Fixed threading and add shutdown control for threads via sleep

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.
 </pre>
 
+### `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.
+
+<pre>
+sleep(50)  # wait for 50 milliseconds
+sleep(1000, _() -> print('Interrupted')) # waits for 1 second, outputs a message when thread is shut down.
+</pre>
+
 ### `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
 </pre>
 
-### `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
-
-<pre>
-sleep(50)
-</pre>
-
 ### `time()`
 
 Returns the number of milliseconds since 'some point', like Java's `System.nanoTime()`, which varies from system to 

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)
         {

src/main/java/carpet/script/CarpetScriptServer.java

@@ -304,7 +304,6 @@ public class CarpetScriptServer
         {
             host.onClose();
         }
-        ThreadValue.shutdown();
     }
 
     public void reload(MinecraftServer server)

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<Value, Value> systemGlobals = new ConcurrentHashMap<>();
     private static final Map<Long, Random> randomizers = new Long2ObjectOpenHashMap<>();
 
+    private final Map<Value, ThreadPoolExecutor> executorServices = new HashMap<>();
+    private final Map<Value, Object> 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<ExecutorService> toStop = new ArrayList<>(executorServices.values());
+        for (ScriptHost uh : userHosts.values()) toStop.addAll(uh.executorServices.values());
+        if (!toStop.isEmpty())
+        {
+            ThreadPoolExecutor shutdownService = (ThreadPoolExecutor) Executors.newCachedThreadPool();
+            List<CompletableFuture<?>> 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)
     {

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);

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
+        });
     }
 }

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<Value> taskFuture;
-    private long id;
+    private final CompletableFuture<Value> taskFuture;
+    private final long id;
     private static long sequence = 0L;
-    private static final Map<Value,ThreadPoolExecutor> executorServices = new HashMap<>();
-    private static final Map<Value, Object> locks = new ConcurrentHashMap<>();
 
     public ThreadValue(Value pool, FunctionValue function, Expression expr, Tokenizer.Token token, Context ctx, List<LazyValue> 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)
     {

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) )