From 1268af4131a4c879c06f6604b5283a7d967b8297 Mon Sep 17 00:00:00 2001 From: gnembon Date: Sat, 24 Oct 2020 22:18:02 -0300 Subject: [PATCH] Fixed threading and add shutdown control for threads via sleep --- docs/scarpet/language/SystemFunctions.md | 43 +++++--- .../java/carpet/script/CarpetScriptHost.java | 1 + .../carpet/script/CarpetScriptServer.java | 1 - src/main/java/carpet/script/ScriptHost.java | 52 ++++++++- src/main/java/carpet/script/language/Sys.java | 12 +- .../carpet/script/language/Threading.java | 43 ++++++-- .../java/carpet/script/value/ThreadValue.java | 104 +++++++----------- .../resources/assets/carpet/scripts/camera.sc | 13 ++- 8 files changed, 162 insertions(+), 107 deletions(-) diff --git a/docs/scarpet/language/SystemFunctions.md b/docs/scarpet/language/SystemFunctions.md index e9e2b8df..88f5af55 100644 --- a/docs/scarpet/language/SystemFunctions.md +++ b/docs/scarpet/language/SystemFunctions.md @@ -81,15 +81,15 @@ str('player at: %d %d %d',pos(player())) => 'player at: 567, -2423, 124' * * * ## Threading and Parallel Execution -Scarpet allows to run threads of execution in parallel to the main script execution thread. In Minecraft, the main -thread scripts are executed on the main server thread. Since Minecraft is inherently NOT thread safe, it is not that -beneficial to parallel execution in order to access world resources faster. Both `GetBlockState` and `setBlockState` +Scarpet allows to run threads of execution in parallel to the main script execution thread. In Minecraft, apps +are executed on the main server thread. Since Minecraft is inherently NOT thread safe, it is not that +beneficial to parallel execution in order to access world resources faster. Both `getBlockState` and `setBlockState` are not thread safe and require the execution to park on the server thread, where these requests can be executed in -the off-tick time in between ticks that didn't take 50ms. There are however benefits of running things in parallel, +the off-tick time in between ticks that didn't take all 50ms. There are however benefits of running things in parallel, like fine time control not relying on the tick clock, or running things independent on each other. You can still run your actions on tick-by-tick basis, either taking control of the execution using `game_tick()` API function -(nasty solution), or scheduling tick using `schedule()` function (much nicer solution), but threading often gives -the neatest solution to solve problems in parallel +(nasty solution), or scheduling tick using `schedule()` function (preffered solution), but threading gives much more control +on the timings without impacting the main game and is the only solution to solve problems in parallel (see [scarpet camera](/src/main/resources/assets/carpet/scripts/camera.sc)). Due to limitations with the game, there are some limits to the threading as well. You cannot for @@ -97,9 +97,11 @@ instance `join_task()` at all from the main script and server thread, because an function that require any world access, will require to park and join on the main thread to get world access, meaning that calling join on that task would inevitably lead to a typical deadlock. You can still join tasks from other threads, just because the only possibility of a deadlock in this case would come explicitly from your -bad code, not the internal world access behaviour. Some things tough like player or entity manipulation, can be +bad code, not the internal world access behaviour. Some things tough like players or entities manipulation, can be effectively parallelized. +If the app is shutting down, creating new tasks via `task` + ### `task(function, ... args)`, `task_thread(executor, function, ... args)` Creates and runs a parallel task, returning the handle to the task object. Task will return the return value of the @@ -118,6 +120,24 @@ a = 3; task_thread('temp', _(outer(a), b) -> foo(a,b), 5) => Another example of running the same thing passing arguments using closure over anonymous function as well as passing a parameter. +### `sleep()` `sleep(timeout)`, `sleep(timeout, close_call, ... args)` + + +Halts the execution of the thread (or the game itself, if run not as a part of a task) for `expr` milliseconds. +It checks for interrupted execution, in that case exits the thread (or the entire program, if not run on a thread) in case the app +is being stopped/removed. If the closing call is specified, executes the following function when a shutdown signal is triggered. +If run on the main thread (not as a taks) the call may only be invocated when the entire game shuts down, so close call only +makes sense for threads. For programs use `__on_close()` handler. + +Since `close_call` is executed after app shutdown is initiated, you won't be able to create new tasks in that block. Threads +should periodically call `sleep` to ensure all app tasks will finish when the app is closing or right after, but the app engine +will not forcefully remove your running tasks so the tasks themselves need to properly react to the closing request. + +
+sleep(50)  # wait for 50 milliseconds
+sleep(1000, _() -> print('Interrupted')) # waits for 1 second, outputs a message when thread is shut down.
+
+ ### `task_count(executor?)` If no argument provided, returns total number of tasks being executed in parallel at this moment using scarpet @@ -255,15 +275,6 @@ print('pi = '+pi) => prints: pi = 3.141592653589793 print(str('pi = %.2f',pi)) => prints: pi = 3.14 -### `sleep(expr)` - -Halts the execution of the program (and the game itself) for `expr` milliseconds. All in all, its better to -use `game_tick(expr)` to let the game do its job while the program waits - -
-sleep(50)
-
- ### `time()` Returns the number of milliseconds since 'some point', like Java's `System.nanoTime()`, which varies from system to diff --git a/src/main/java/carpet/script/CarpetScriptHost.java b/src/main/java/carpet/script/CarpetScriptHost.java index 29c17140..cdfd7b74 100644 --- a/src/main/java/carpet/script/CarpetScriptHost.java +++ b/src/main/java/carpet/script/CarpetScriptHost.java @@ -337,6 +337,7 @@ public class CarpetScriptHost extends ScriptHost @Override public void onClose() { + inTermination = true; FunctionValue closing = getFunction("__on_close"); if (closing != null) { diff --git a/src/main/java/carpet/script/CarpetScriptServer.java b/src/main/java/carpet/script/CarpetScriptServer.java index f7015c30..73bd60c4 100644 --- a/src/main/java/carpet/script/CarpetScriptServer.java +++ b/src/main/java/carpet/script/CarpetScriptServer.java @@ -304,7 +304,6 @@ public class CarpetScriptServer { host.onClose(); } - ThreadValue.shutdown(); } public void reload(MinecraftServer server) diff --git a/src/main/java/carpet/script/ScriptHost.java b/src/main/java/carpet/script/ScriptHost.java index 536bd4fb..27326e06 100644 --- a/src/main/java/carpet/script/ScriptHost.java +++ b/src/main/java/carpet/script/ScriptHost.java @@ -9,14 +9,20 @@ import carpet.script.value.Value; import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; import it.unimi.dsi.fastutil.objects.Object2ObjectOpenHashMap; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Random; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadPoolExecutor; import java.util.function.Predicate; +import java.util.stream.Collectors; import java.util.stream.Stream; public abstract class ScriptHost @@ -24,6 +30,10 @@ public abstract class ScriptHost public static Map systemGlobals = new ConcurrentHashMap<>(); private static final Map randomizers = new Long2ObjectOpenHashMap<>(); + private final Map executorServices = new HashMap<>(); + private final Map locks = new ConcurrentHashMap<>(); + protected boolean inTermination = false; + public Random getRandom(long aLong) { if (randomizers.size() > 1024) @@ -31,7 +41,6 @@ public abstract class ScriptHost return randomizers.computeIfAbsent(aLong, Random::new); } - public static class ModuleData { Module parent; @@ -325,7 +334,46 @@ public abstract class ScriptHost protected abstract ScriptHost duplicate(); - public void onClose() { } + public Object getLock(Value name) + { + return locks.computeIfAbsent(name, (n) -> new Object()); + } + + public ThreadPoolExecutor getExecutor(Value pool) + { + if (inTermination) return null; + return executorServices.computeIfAbsent(pool, (v) -> (ThreadPoolExecutor) Executors.newCachedThreadPool()); + } + + public int taskCount() + { + return executorServices.values().stream().map(ThreadPoolExecutor::getActiveCount).reduce(0, Integer::sum); + } + public int taskCount(Value pool) + { + if (executorServices.containsKey(pool)) + { + return executorServices.get(pool).getActiveCount(); + } + return 0; + } + + public void onClose() + { + inTermination = true; + List toStop = new ArrayList<>(executorServices.values()); + for (ScriptHost uh : userHosts.values()) toStop.addAll(uh.executorServices.values()); + if (!toStop.isEmpty()) + { + ThreadPoolExecutor shutdownService = (ThreadPoolExecutor) Executors.newCachedThreadPool(); + List> terminators = toStop.stream().map(exec -> + CompletableFuture.supplyAsync(() -> { exec.shutdown(); exec.shutdownNow(); return null; }, shutdownService)).collect(Collectors.toList() + ); + terminators.forEach(CompletableFuture::join); + shutdownService.shutdown(); + shutdownService.shutdownNow(); + } + } public void setPerPlayer(boolean isPerUser) { diff --git a/src/main/java/carpet/script/language/Sys.java b/src/main/java/carpet/script/language/Sys.java index 22b583a1..2f21cb4e 100644 --- a/src/main/java/carpet/script/language/Sys.java +++ b/src/main/java/carpet/script/language/Sys.java @@ -294,17 +294,7 @@ public class Sys { System.out.println(v.getString()); return v; // pass through for variables }); - expression.addUnaryFunction("sleep", (v) -> - { - long time = NumericValue.asNumber(v).getLong(); - try - { - Thread.sleep(time); - Thread.yield(); - } - catch (InterruptedException ignored) { } - return v; // pass through for variables - }); + expression.addLazyFunction("time", 0, (c, t, lv) -> { Value time = new NumericValue((System.nanoTime() / 1000) / 1000.0); diff --git a/src/main/java/carpet/script/language/Threading.java b/src/main/java/carpet/script/language/Threading.java index 938c72ca..15383e7d 100644 --- a/src/main/java/carpet/script/language/Threading.java +++ b/src/main/java/carpet/script/language/Threading.java @@ -2,7 +2,9 @@ package carpet.script.language; import carpet.script.Expression; import carpet.script.argument.FunctionArgument; +import carpet.script.exception.ExitStatement; import carpet.script.exception.InternalExpressionException; +import carpet.script.value.FunctionValue; import carpet.script.value.NumericValue; import carpet.script.value.ThreadValue; import carpet.script.value.Value; @@ -33,13 +35,12 @@ public class Threading }); - expression.addFunction("task_count", (lv) -> + expression.addLazyFunction("task_count", -1, (c, t, lv) -> { - if (lv.size() > 0) - { - return new NumericValue(ThreadValue.taskCount(lv.get(0))); - } - return new NumericValue(ThreadValue.taskCount()); + Value ret = (lv.size() > 0)? + new NumericValue(c.host.taskCount(lv.get(0).evalValue(c))): + new NumericValue(c.host.taskCount()); + return (cc, tt) -> ret; }); expression.addUnaryFunction("task_value", (v) -> @@ -80,11 +81,39 @@ public class Threading lockValue = lv.get(0).evalValue(c); ind = 1; } - synchronized (ThreadValue.getLock(lockValue)) + synchronized (c.host.getLock(lockValue)) { Value ret = lv.get(ind).evalValue(c, t); return (_c, _t) -> ret; } }); + + expression.addLazyFunctionWithDelegation("sleep", -1, (c, t, expr, tok, lv) -> + { + long time = lv.isEmpty()?0L:NumericValue.asNumber(lv.get(0).evalValue(c)).getLong(); + boolean interrupted = false; + try + { + if (Thread.interrupted()) interrupted = true; + if (time > 0) Thread.sleep(time); + Thread.yield(); + } + catch (InterruptedException ignored) + { + interrupted = true; + } + if (interrupted) + { + Value exceptionally = Value.NULL; + if (lv.size() > 1) + { + FunctionArgument functionArgument = FunctionArgument.findIn(c, expression.module, lv, 1, false); + FunctionValue fun = functionArgument.function; + exceptionally = fun.callInContext(expr, c, t, fun.getExpression(), fun.getToken(), functionArgument.args).evalValue(c); + } + throw new ExitStatement(exceptionally); + } + return (cc, tt) -> new NumericValue(time); // pass through for variables + }); } } diff --git a/src/main/java/carpet/script/value/ThreadValue.java b/src/main/java/carpet/script/value/ThreadValue.java index 45cecceb..11be5b93 100644 --- a/src/main/java/carpet/script/value/ThreadValue.java +++ b/src/main/java/carpet/script/value/ThreadValue.java @@ -4,47 +4,55 @@ import carpet.script.Context; import carpet.script.Expression; import carpet.script.LazyValue; import carpet.script.Tokenizer; +import carpet.script.exception.ExitStatement; import carpet.script.exception.ExpressionException; import carpet.script.exception.InternalExpressionException; import net.minecraft.nbt.Tag; -import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; public class ThreadValue extends Value { - private CompletableFuture taskFuture; - private long id; + private final CompletableFuture taskFuture; + private final long id; private static long sequence = 0L; - private static final Map executorServices = new HashMap<>(); - private static final Map locks = new ConcurrentHashMap<>(); public ThreadValue(Value pool, FunctionValue function, Expression expr, Tokenizer.Token token, Context ctx, List args) { id = sequence++; - taskFuture = CompletableFuture.supplyAsync( - () -> - { - try - { - return function.lazyEval(ctx, Context.NONE, expr, token, args).evalValue(ctx); - } - catch (ExpressionException exc) - { - ctx.host.handleExpressionException("Thread failed", exc); - return Value.NULL; - } - }, - executorServices.computeIfAbsent(pool, (v) -> (ThreadPoolExecutor)Executors.newCachedThreadPool()) - ); + ExecutorService executor = ctx.host.getExecutor(pool); + if (executor == null) + { + // app is shutting down - no more threads can be spawned. + taskFuture = CompletableFuture.completedFuture(Value.NULL); + } + else + { + taskFuture = CompletableFuture.supplyAsync( + () -> + { + try + { + return function.lazyEval(ctx, Context.NONE, expr, token, args).evalValue(ctx); + } + catch (ExitStatement exit) + { + // app stopped + return exit.retval; + } + catch (ExpressionException exc) + { + ctx.host.handleExpressionException("Thread failed", exc); + return Value.NULL; + } + + }, + ctx.host.getExecutor(pool) + ); + } Thread.yield(); } @@ -71,6 +79,11 @@ public class ThreadValue extends Value { return taskFuture.get(); } + catch (ExitStatement exit) + { + taskFuture.complete(exit.retval); + return exit.retval; + } catch (InterruptedException | ExecutionException e) { return Value.NULL; @@ -104,47 +117,6 @@ public class ThreadValue extends Value return Long.hashCode(id); } - public static int taskCount() - { - return executorServices.values().stream().map(ThreadPoolExecutor::getActiveCount).reduce(0, Integer::sum); - } - - public static int taskCount(Value pool) - { - if (executorServices.containsKey(pool)) - { - return executorServices.get(pool).getActiveCount(); - } - return 0; - } - - public static Object getLock(Value name) - { - return locks.computeIfAbsent(name, (n) -> new Object()); - } - - - public static void shutdown() - { - for (ExecutorService exec : executorServices.values()) - { - exec.shutdown(); - try - { - exec.awaitTermination(5000, TimeUnit.MILLISECONDS); - } - catch (InterruptedException ignored) - { - } - finally - { - exec.shutdownNow(); - } - } - executorServices.clear(); - locks.clear(); - } - @Override public Tag toTag(boolean force) { diff --git a/src/main/resources/assets/carpet/scripts/camera.sc b/src/main/resources/assets/carpet/scripts/camera.sc index 399f0003..28b18fc3 100644 --- a/src/main/resources/assets/carpet/scripts/camera.sc +++ b/src/main/resources/assets/carpet/scripts/camera.sc @@ -458,6 +458,12 @@ show() -> task( _() -> ( global_markers = __create_markers(); + on_close = ( _() -> ( + for(global_markers, modify(_,'remove')); + global_markers = null; + global_showing_path = false; + )); + loop(7200, if(!global_showing_path, break()); __get_player(); @@ -467,11 +473,9 @@ show() -> global_markers = __create_markers(); ); __show_path_tick(); - sleep(100); + sleep(100, on_close); ); - for(global_markers, modify(_,'remove')); - global_markers = null; - global_showing_path = false; + call(on_close); )); null; ); @@ -523,6 +527,7 @@ play() -> modify(p, 'location', v); point += 1; end_time = time(); + sleep(); if (global_prefer_sync, should_be = very_start + mspt*point; if (end_time < should_be, sleep(should_be-end_time) )