diff --git a/module/minecraft/minecraft-kether/build.gradle.kts b/module/minecraft/minecraft-kether/build.gradle.kts index d4fcec6c9..417e96359 100644 --- a/module/minecraft/minecraft-kether/build.gradle.kts +++ b/module/minecraft/minecraft-kether/build.gradle.kts @@ -4,8 +4,10 @@ import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar dependencies { compileOnly(project(":common")) + testImplementation(project(":common")) compileOnly(project(":common-env")) compileOnly(project(":common-util")) + testImplementation(project(":common-util")) compileOnly(project(":common-legacy-api")) compileOnly(project(":common-platform-api")) compileOnly(project(":module:minecraft:minecraft-chat")) diff --git a/module/minecraft/minecraft-kether/src/main/java/taboolib/library/kether/AbstractQuestContext.java b/module/minecraft/minecraft-kether/src/main/java/taboolib/library/kether/AbstractQuestContext.java index a786f2f45..fac45b22e 100644 --- a/module/minecraft/minecraft-kether/src/main/java/taboolib/library/kether/AbstractQuestContext.java +++ b/module/minecraft/minecraft-kether/src/main/java/taboolib/library/kether/AbstractQuestContext.java @@ -4,6 +4,7 @@ import org.jetbrains.annotations.NotNull; import java.util.*; +import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.Executor; @@ -15,8 +16,8 @@ public abstract class AbstractQuestContext> im protected final Frame rootFrame; protected final Quest quest; protected final QuestExecutor executor; - protected ExitStatus exitStatus; - protected CompletableFuture future; + protected volatile ExitStatus exitStatus; + protected volatile CompletableFuture future; protected AbstractQuestContext(QuestService service, Quest quest, String playerIdentifier) { this.service = service; @@ -61,18 +62,31 @@ public Frame rootFrame() { } @Override - public CompletableFuture runActions() { + public synchronized CompletableFuture runActions() { Preconditions.checkState(future == null, "already running"); - return future = rootFrame.run().thenApply(o -> { - if (this.exitStatus == null) { - this.exitStatus = ExitStatus.success(); + CompletableFuture frameFuture = rootFrame.run(); + CompletableFuture contextFuture = new CompletableFuture<>(); + frameFuture.whenComplete((result, ex) -> { + if (ex != null) { + completeFailure(contextFuture, ex); + } else { + if (this.exitStatus == null) { + this.exitStatus = ExitStatus.success(); + } + contextFuture.complete(result); + } + }); + contextFuture.whenComplete((result, ex) -> { + if (contextFuture.isCancelled()) { + frameFuture.cancel(false); } - return o; }); + this.future = contextFuture; + return contextFuture; } @Override - public void terminate() { + public synchronized void terminate() { this.rootFrame.close(); if (future != null) { future.completeExceptionally(new QuestCloseException()); @@ -80,6 +94,18 @@ public void terminate() { } } + private static void completeFailure(CompletableFuture future, Throwable throwable) { + Throwable cause = throwable; + while (cause instanceof CompletionException && cause.getCause() != null) { + cause = cause.getCause(); + } + if (cause instanceof CancellationException) { + future.cancel(false); + } else { + future.completeExceptionally(cause); + } + } + public static class QuestExecutor implements Executor { private final AbstractQuestContext questContext; @@ -104,7 +130,7 @@ public static abstract class AbstractFrame implements Frame { protected final List frames; protected final VarTable varTable; protected final QuestContext questContext; - protected CompletableFuture future; + protected volatile CompletableFuture future; protected final Deque closeables = new LinkedBlockingDeque<>(); public AbstractFrame(Frame parent, List frames, VarTable varTable, QuestContext questContext) { @@ -162,12 +188,14 @@ public T addClosable(T closeable) { @Override public void close() { - if (this.future == null) return; + CompletableFuture runningFuture = this.future; + if (runningFuture == null) return; + this.future = null; for (Frame frame : this.frames) { frame.close(); } this.cleanup(); - this.future = null; + runningFuture.completeExceptionally(new QuestCloseException()); } @Override @@ -191,6 +219,7 @@ public static class SimpleNamedFrame extends AbstractFrame { private final String name; private Quest.Block block, next; private int sp = -1, np = -1; + private volatile CompletableFuture runningAction; public SimpleNamedFrame(Frame parent, List frames, VarTable varTable, String name, QuestContext questContext) { super(parent, frames, varTable, questContext); @@ -235,36 +264,100 @@ public void setNext(@NotNull Quest.Block block) { np = 0; } + @Override + public synchronized void close() { + CompletableFuture actionFuture = this.runningAction; + this.runningAction = null; + super.close(); + if (actionFuture != null) { + actionFuture.cancel(false); + } + } + @Override @SuppressWarnings("unchecked") - public CompletableFuture run() { + public synchronized CompletableFuture run() { Preconditions.checkState(this.future == null, "already running"); varTable.initialize(this); future = new CompletableFuture<>(); - process(future); - return (CompletableFuture) future; + CompletableFuture resultFuture = future; + resultFuture.whenComplete((result, ex) -> { + if (resultFuture.isCancelled()) { + this.close(); + } + }); + process(null); + return (CompletableFuture) resultFuture; } - @SuppressWarnings("unchecked") - private void process(CompletableFuture future) { + private synchronized void process(CompletableFuture previousFuture) { + CompletableFuture resultFuture = this.future; + if (resultFuture == null || resultFuture.isDone()) { + return; + } while (!context().getExitStatus().isPresent()) { this.cleanup(); this.frames.removeIf(Frame::isDone); Optional> optional = nextAction(); - if (optional.isPresent()) { - ParsedAction action = optional.get(); - CompletableFuture newFuture = action.process(this); - if (!newFuture.isDone()) { - newFuture.thenRun(() -> this.process(newFuture)); - return; - } else { - future = newFuture; - } - } else { - ((CompletableFuture) this.future).complete(future != null && future.isDone() ? future.join() : null); + if (!optional.isPresent()) { + completeResult(resultFuture, previousFuture); + return; + } + ParsedAction action = optional.get(); + CompletableFuture actionFuture; + try { + actionFuture = Objects.requireNonNull(action.process(this), "Quest action returned null future: " + action); + } catch (Throwable ex) { + fail(resultFuture, ex); return; } + this.runningAction = actionFuture; + if (!actionFuture.isDone()) { + actionFuture.whenComplete((result, ex) -> resume(resultFuture, actionFuture, ex)); + return; + } + this.runningAction = null; + if (actionFuture.isCancelled()) { + resultFuture.cancel(false); + return; + } + try { + actionFuture.join(); + } catch (Throwable ex) { + fail(resultFuture, ex); + return; + } + previousFuture = actionFuture; } + this.cleanup(); + this.frames.removeIf(Frame::isDone); + completeResult(resultFuture, previousFuture); + } + + private synchronized void resume(CompletableFuture resultFuture, CompletableFuture actionFuture, Throwable throwable) { + if (this.runningAction == actionFuture) { + this.runningAction = null; + } + if (this.future != resultFuture || resultFuture.isDone()) { + return; + } + if (throwable != null) { + fail(resultFuture, throwable); + } else { + process(actionFuture); + } + } + + private void fail(CompletableFuture resultFuture, Throwable throwable) { + this.cleanup(); + this.frames.removeIf(Frame::isDone); + completeFailure(resultFuture, throwable); + } + + @SuppressWarnings("unchecked") + private void completeResult(CompletableFuture resultFuture, CompletableFuture previousFuture) { + Object result = previousFuture != null ? previousFuture.getNow(null) : null; + ((CompletableFuture) resultFuture).complete(result); } private Optional> nextAction() { @@ -309,10 +402,17 @@ public void setNext(@NotNull Quest.Block block) { @Override @SuppressWarnings("unchecked") - public CompletableFuture run() { + public synchronized CompletableFuture run() { Preconditions.checkState(this.future == null, "already running"); this.varTable.initialize(this); - return (CompletableFuture) (this.future = this.action.process(this)); + try { + this.future = Objects.requireNonNull(this.action.process(this), "Quest action returned null future: " + action); + } catch (Throwable ex) { + CompletableFuture failed = new CompletableFuture<>(); + completeFailure(failed, ex); + this.future = failed; + } + return (CompletableFuture) this.future; } } diff --git a/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/RemoteQuestReader.kt b/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/RemoteQuestReader.kt index 92fe7aca4..d2529dc4d 100644 --- a/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/RemoteQuestReader.kt +++ b/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/RemoteQuestReader.kt @@ -12,44 +12,54 @@ import taboolib.library.kether.QuestReader @Suppress("UNCHECKED_CAST") class RemoteQuestReader(val remote: OpenContainer, val source: Any) : QuestReader { + @Synchronized override fun peek(): Char { return source.invokeMethod("peek", remap = false)!! } + @Synchronized override fun peek(n: Int): Char { return peekIntMethod[source].invoke(source, n) as Char } + @Synchronized override fun getIndex(): Int { return source.invokeMethod("getIndex", remap = false)!! } + @Synchronized override fun getMark(): Int { return source.invokeMethod("getMark", remap = false)!! } + @Synchronized override fun hasNext(): Boolean { return source.invokeMethod("hasNext", remap = false)!! } + @Synchronized override fun nextToken(): String { return source.invokeMethod("nextToken", remap = false)!! } + @Synchronized override fun mark() { source.invokeMethod("mark", remap = false) } + @Synchronized override fun reset() { source.invokeMethod("reset", remap = false) } + @Synchronized override fun nextAction(): ParsedAction { val action = source.invokeMethod("nextAction", remap = false)!! val questAction = RemoteQuestAction(remote, action.getProperty("action", remap = false)!!) return ParsedAction(questAction, action.getProperty>("properties", remap = false)!!) } + @Synchronized override fun nextAction(namespace: String?): ParsedAction { return try { val action = nextActionStringMethod[source].invoke(source, namespace)!! @@ -60,6 +70,7 @@ class RemoteQuestReader(val remote: OpenContainer, val source: Any) : QuestReade } } + @Synchronized override fun expect(value: String) { expectMethod[source].invoke(source, value) } diff --git a/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/action/game/bukkit/ActionScoreboard.kt b/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/action/game/bukkit/ActionScoreboard.kt index c6fee83f5..7264c67c8 100644 --- a/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/action/game/bukkit/ActionScoreboard.kt +++ b/module/minecraft/minecraft-kether/src/main/kotlin/taboolib/module/kether/action/game/bukkit/ActionScoreboard.kt @@ -4,9 +4,14 @@ import org.bukkit.entity.Player import taboolib.common.Inject import taboolib.common.platform.Platform import taboolib.common.platform.PlatformSide +import taboolib.common.platform.function.submit import taboolib.common.util.asList import taboolib.module.kether.* import taboolib.module.nms.sendScoreboard +import java.util.concurrent.CancellationException +import java.util.concurrent.CompletableFuture +import java.util.concurrent.CompletionException +import java.util.concurrent.atomic.AtomicReference @Inject @PlatformSide(Platform.BUKKIT) @@ -15,16 +20,84 @@ object ActionScoreboard { @KetherParser(["scoreboard"]) fun actionScoreboard() = scriptParser { val value = it.nextParsedAction() - actionNow { - run(value).thenAccept { o -> - val viewer = player().cast() - if (o == null) { - viewer.sendScoreboard() - } else { - val body = if (o is Collection<*> || o is Array<*>) o.asList() else o.toString().trimIndent().lines() - viewer.sendScoreboard(body[0], *body.filterIndexed { index, _ -> index > 0 }.toTypedArray()) + actionTake { + val viewer = player().cast() + val result = CompletableFuture() + val updateFuture = AtomicReference?>() + val contentFuture = run(value) + contentFuture.whenComplete { content, ex -> + if (ex != null) { + completeFailure(result, ex) + } else if (!result.isDone) { + val scoreboardFuture = updateScoreboard(viewer, content) + updateFuture.set(scoreboardFuture) + if (result.isCancelled) { + scoreboardFuture.cancel(false) + } else { + scoreboardFuture.whenComplete { _, updateEx -> + if (updateEx != null) { + completeFailure(result, updateEx) + } else { + result.complete(null) + } + } + } } } + result.whenComplete { _, _ -> + if (result.isCancelled) { + contentFuture.cancel(false) + updateFuture.get()?.cancel(false) + } + } + result + } + } + + private fun completeFailure(future: CompletableFuture<*>, throwable: Throwable) { + var cause = throwable + while (cause is CompletionException) { + val nested = cause.cause ?: break + cause = nested + } + if (cause is CancellationException) { + future.cancel(false) + } else { + future.completeExceptionally(cause) + } + } + + private fun updateScoreboard(viewer: Player, content: Any?): CompletableFuture { + val future = CompletableFuture() + try { + val task = submit { + if (future.isCancelled) { + return@submit + } + try { + val body = when (content) { + null -> emptyList() + is Collection<*>, is Array<*> -> content.asList() + else -> content.toString().trimIndent().lines() + } + if (body.isEmpty()) { + viewer.sendScoreboard() + } else { + viewer.sendScoreboard(body.first(), *body.drop(1).toTypedArray()) + } + future.complete(null) + } catch (ex: Throwable) { + future.completeExceptionally(ex) + } + } + future.whenComplete { _, _ -> + if (future.isCancelled) { + task.cancel() + } + } + } catch (ex: Throwable) { + future.completeExceptionally(ex) } + return future } -} \ No newline at end of file +} diff --git a/module/minecraft/minecraft-kether/src/test/java/taboolib/library/kether/AbstractQuestContextTest.java b/module/minecraft/minecraft-kether/src/test/java/taboolib/library/kether/AbstractQuestContextTest.java new file mode 100644 index 000000000..934e19ce9 --- /dev/null +++ b/module/minecraft/minecraft-kether/src/test/java/taboolib/library/kether/AbstractQuestContextTest.java @@ -0,0 +1,219 @@ +package taboolib.library.kether; + +import org.jetbrains.annotations.NotNull; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.Executor; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class AbstractQuestContextTest { + + @Test + void asynchronousActionFailureCompletesContextExceptionally() { + CompletableFuture actionFuture = new CompletableFuture<>(); + IllegalStateException failure = new IllegalStateException("boom"); + TestQuestContext context = context(action(frame -> actionFuture)); + + CompletableFuture result = context.runActions(); + actionFuture.completeExceptionally(failure); + + CompletionException thrown = assertThrows(CompletionException.class, result::join); + assertSame(failure, thrown.getCause()); + } + + @Test + void completedExceptionalActionDoesNotEscapeRunActions() { + IllegalStateException failure = new IllegalStateException("boom"); + CompletableFuture failed = new CompletableFuture<>(); + failed.completeExceptionally(failure); + TestQuestContext context = context(action(frame -> failed)); + + CompletableFuture result = assertDoesNotThrow(context::runActions); + + CompletionException thrown = assertThrows(CompletionException.class, result::join); + assertSame(failure, thrown.getCause()); + } + + @Test + void synchronousActionFailureStopsFollowingActions() { + IllegalStateException failure = new IllegalStateException("boom"); + AtomicInteger followingRuns = new AtomicInteger(); + TestQuestContext context = context( + action(frame -> { + throw failure; + }), + action(frame -> { + followingRuns.incrementAndGet(); + return CompletableFuture.completedFuture(null); + }) + ); + + CompletableFuture result = assertDoesNotThrow(context::runActions); + + CompletionException thrown = assertThrows(CompletionException.class, result::join); + assertSame(failure, thrown.getCause()); + assertEquals(0, followingRuns.get()); + } + + @Test + void actionFrameConvertsSynchronousFailureToFuture() { + IllegalStateException failure = new IllegalStateException("boom"); + TestQuestContext context = context(); + QuestContext.Frame frame = context.rootFrame().newFrame(action(ignored -> { + throw failure; + })); + + CompletableFuture result = assertDoesNotThrow(() -> { + return frame.run(); + }); + + CompletionException thrown = assertThrows(CompletionException.class, result::join); + assertSame(failure, thrown.getCause()); + } + + @Test + void exitStatusCompletesWithLastActionValue() { + AtomicInteger closes = new AtomicInteger(); + TestQuestContext context = context(action(frame -> { + frame.addClosable(closes::incrementAndGet); + frame.context().setExitStatus(ExitStatus.success()); + return CompletableFuture.completedFuture(7); + })); + + assertEquals(7, context.runActions().join()); + assertEquals(1, closes.get()); + } + + @Test + void cancellingContextCancelsRunningAction() { + CompletableFuture actionFuture = new CompletableFuture<>(); + TestQuestContext context = context(action(frame -> actionFuture)); + CompletableFuture result = context.runActions(); + + assertTrue(result.cancel(false)); + + assertTrue(actionFuture.isCancelled()); + assertTrue(result.isCancelled()); + } + + @Test + void terminatingContextClosesFrameAndRunningAction() { + CompletableFuture actionFuture = new CompletableFuture<>(); + TestQuestContext context = context(action(frame -> actionFuture)); + CompletableFuture result = context.runActions(); + + context.terminate(); + + assertTrue(actionFuture.isCancelled()); + assertTrue(result.isCompletedExceptionally()); + assertFalse(result.isCancelled()); + } + + @SafeVarargs + private final TestQuestContext context(ParsedAction... actions) { + return new TestQuestContext(new TestQuest(Arrays.asList(actions))); + } + + private ParsedAction action(ActionProcessor processor) { + return new ParsedAction<>(new QuestAction() { + @Override + public CompletableFuture process(@NotNull QuestContext.Frame frame) { + return processor.process(frame); + } + }); + } + + private interface ActionProcessor { + + CompletableFuture process(QuestContext.Frame frame); + } + + private static class TestQuestContext extends AbstractQuestContext { + + TestQuestContext(Quest quest) { + super(null, quest, "test"); + } + + @Override + protected Executor createExecutor() { + return Runnable::run; + } + } + + private static class TestQuest implements Quest { + + private final Map blocks; + + TestQuest(List> actions) { + Map values = new LinkedHashMap<>(); + values.put(QuestContext.BASE_BLOCK, new TestBlock(QuestContext.BASE_BLOCK, actions)); + this.blocks = Collections.unmodifiableMap(values); + } + + @Override + public String getId() { + return "test"; + } + + @Override + public Optional getBlock(@NotNull String label) { + return Optional.ofNullable(blocks.get(label)); + } + + @Override + public Map getBlocks() { + return blocks; + } + + @Override + public Optional blockOf(@NotNull ParsedAction action) { + return blocks.values().stream().filter(block -> block.indexOf(action) >= 0).findFirst(); + } + } + + private static class TestBlock implements Quest.Block { + + private final String label; + private final List> actions; + + TestBlock(String label, List> actions) { + this.label = label; + this.actions = actions; + } + + @Override + public String getLabel() { + return label; + } + + @Override + public List> getActions() { + return actions; + } + + @Override + public int indexOf(@NotNull ParsedAction action) { + return actions.indexOf(action); + } + + @Override + public Optional> get(int index) { + return index >= 0 && index < actions.size() ? Optional.of(actions.get(index)) : Optional.empty(); + } + } +} diff --git a/module/minecraft/minecraft-kether/src/test/kotlin/taboolib/module/kether/RemoteQuestReaderTest.kt b/module/minecraft/minecraft-kether/src/test/kotlin/taboolib/module/kether/RemoteQuestReaderTest.kt new file mode 100644 index 000000000..caf844ff1 --- /dev/null +++ b/module/minecraft/minecraft-kether/src/test/kotlin/taboolib/module/kether/RemoteQuestReaderTest.kt @@ -0,0 +1,64 @@ +package taboolib.module.kether + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import taboolib.common.OpenContainer +import taboolib.common.OpenResult +import java.util.concurrent.CountDownLatch +import java.util.concurrent.Executors +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.locks.LockSupport + +class RemoteQuestReaderTest { + + @Test + fun `reader operations are serialized per remote source`() { + val source = ConcurrentReaderSource() + val reader = RemoteQuestReader(TestContainer, source) + val executor = Executors.newFixedThreadPool(8) + val start = CountDownLatch(1) + try { + val tasks = List(32) { + executor.submit { + start.await() + reader.nextToken() + } + } + start.countDown() + tasks.forEach { future -> + assertEquals("token", future.get(5, TimeUnit.SECONDS)) + } + } finally { + executor.shutdownNow() + } + + assertEquals(1, source.maxConcurrentCalls.get()) + } + + private class ConcurrentReaderSource { + + private val activeCalls = AtomicInteger() + val maxConcurrentCalls = AtomicInteger() + + fun nextToken(): String { + val active = activeCalls.incrementAndGet() + maxConcurrentCalls.updateAndGet { current -> maxOf(current, active) } + try { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(2)) + return "token" + } finally { + activeCalls.decrementAndGet() + } + } + } + + private object TestContainer : OpenContainer { + + override fun isValid() = true + + override fun getName() = "test" + + override fun call(name: String, args: Array) = OpenResult.failed() + } +}