lib_snapshot: First good story of interruptable/resumable compute

examples/snapshot_example.metta

@@ -0,0 +1,13 @@
+!(import! &self (library lib_snapshot))
+
+;Functions as usual:
+(= (fib-tr $n $a $b)
+   (if (== $n 0)
+       $a
+       (fib-tr (- $n 1) $b (+ $a $b))))
+
+(= (fib $n)
+   (fib-tr $n 0 1))
+
+;Start computation:
+!(CallWithSnapshots (fib 1000) 300)

lib/lib_snapshot.metta

@@ -0,0 +1,52 @@
+!(import! &self (library lib_import))
+!(import_prolog_functions_from_file (library lib_snapshot.pl) (qsave_program_continuation))
+!(import! &self (library lib_spaces))
+!(import_prolog_function statistics)
+!(import_prolog_function reset)
+
+(= (BudgetLimit) (empty))
+
+(= (BudgetLimitCheck)
+   (if (> (- (statistics inferences) (get-state &initinferences)) (BudgetLimit))
+       (translatePredicate (shift (pay-for-inferences)))
+       True))
+
+(= (ResetBudgetLimit)
+   (progn (change-state! &initinferences (statistics inferences))
+          (change-state! &inferenceslimit $N)))
+
+(= (InjectBudgetLimitCheck)
+   (match &self $func (let (cons = ($head $body)) $func
+                           (if (not (is-member (car-atom $head)
+                                               (quote (BudgetLimit BudgetLimitCheck ResetBudgetLimit InjectBudgetLimitCheck
+                                                       CallWithSnapshotsInternal CallWithSnapshots ResumeComputation))))
+                               (progn (remove-atom &self $func)
+                                      (add-atom &self (= $head (progn (BudgetLimitCheck) $body))))))))
+
+(= (CallWithSnapshotsInternal $predicate $var)
+   (let $cont (reset $predicate $signal)
+        (if (is-var $signal)
+            $var
+            (progn ;metadata with clause refs needs to be dropped:
+                   (translatePredicate (retractall (Predicate (translated_from $1 $2))))
+                   ;save current atom space with computation continuation entry point ResumeComputation:
+                   (qsave_program_continuation "continue.p" $cont $var ResumeComputation)
+                   (progn (ResetBudgetLimit)
+                          ;continue computation:
+                          (CallWithSnapshotsInternal $cont $var))))))
+
+(: CallWithSnapshots (-> Expression Atom Atom))
+(= (CallWithSnapshots $G $Limit)
+   (progn (add-atom &self (= (BudgetLimit) $Limit))
+          (ResetBudgetLimit)
+          (collapse (InjectBudgetLimitCheck))
+          (let $ext (Predicate (union-atom $G ($X)))
+               (CallWithSnapshotsInternal $ext $X))))
+
+;Function to resume computation with "swipl -x continue.p":
+(= (ResumeComputation $cont $var)
+   (progn (ResetBudgetLimit)
+          ;Continue computation:
+          (let $result (CallWithSnapshotsInternal $cont $var)
+               (println! $result))
+          (translatePredicate (halt))))

lib/lib_snapshot.pl

@@ -0,0 +1,74 @@
+%^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^%
+%                         Continuation (de)serializer by Siyuan Chen:                                              %
+% https://swi prolog.discourse.group/t/is there a way to serialize continuation or program execution state/5548/19 %
+%vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv%
+
+%Serialize continuation:
+serialize(Term, STerm), \+ compound(Term) => STerm = Term.
+serialize(Term, STerm), is_list(Term) => maplist([In, Out] >> serialize(In, Out), Term, STerm).
+serialize(Term, STerm), Term =.. ['$cont$', Module, Clause, PC | Args] => maplist([In, Out] >> serialize(In, Out), Args, SArgs),
+                                                                          nth_clause(Pred, Index, Clause),
+                                                                          STerm =.. ['$cont$', Module, '$CLAUSE'(Pred, Index), PC | SArgs].
+serialize(Term, STerm), Term =.. [Name | Args] => maplist([In, Out] >> serialize(In, Out), Args, SArgs),
+                                                  STerm =.. [Name | SArgs].
+
+%Deserialize continuaton:
+deserialize(STerm, Term), \+ compound(STerm) => Term = STerm.
+deserialize(STerm, Term), is_list(STerm) => maplist([In, Out] >> deserialize(In, Out), STerm, Term).
+deserialize(STerm, Term), STerm =.. ['$cont$', Module, '$CLAUSE'(Pred, Index), PC | SArgs] => maplist([In, Out] >> deserialize(In, Out), SArgs, Args),
+                                                                                              nth_clause(Pred, Index, Clause),
+                                                                                              Term =.. ['$cont$', Module, Clause, PC | Args].
+deserialize(STerm, Term), STerm =.. [Name | SArgs] => maplist([In, Out] >> deserialize(In, Out), SArgs, Args),
+                                                      Term =.. [Name | Args].
+
+%^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^%
+%                         PeTTa compute state save and restore via file by Patrick Hammer:                         %
+%vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv%
+
+atomic_write_term(File, Term, Options) :- string_concat(File, ".tmp", Tmp),
+                                          open(Tmp, write, Out),
+                                          write_term(Out, Term, Options),
+                                          flush_output(Out),
+                                          close(Out),
+                                          rename_file(Tmp, File).
+
+%qsave_program but with saved continuation:
+qsave_program_continuation(Name, Cont, Var, Resume, true) :- string_concat(Name, ".continuation", ContFile),
+                                                             string_concat(Name, ".outputvariable", VarFile),
+                                                             string_concat(Name, ".commit", CommitFile),
+                                                             string_concat(Name, ".commit.new", NewCommitFile),
+                                                             %Serialize and save continuation
+                                                             serialize(Cont, SCont),
+                                                             atomic_write_term(ContFile, SCont, [quoted(true), fullstop(true), nl(true)]),
+                                                             %Save output variable index:
+                                                             term_variables(Cont, Vars),
+                                                             nth0(Index, Vars, V),
+                                                             V == Var,
+                                                             atomic_write_term(VarFile, Index, [fullstop(true), nl(true)]),
+                                                             atomic_write_term(NewCommitFile, ok, [fullstop(true), nl(true)]),
+                                                             rename_file(NewCommitFile, CommitFile),
+                                                             %Save clauses:
+                                                             string_concat(Name, ".new", TmpExe),
+                                                             qsave_program(TmpExe, [goal(load_saved_continuation(Name, Resume))]),
+                                                             rename_file(TmpExe, Name).
+
+%Restoring continuation after qsaved program gets re-invoked:
+load_saved_continuation(Name, Resume) :- string_concat(Name, ".continuation", ContFile),
+                                         string_concat(Name, ".outputvariable", VarFile),
+                                         string_concat(Name, ".commit", CommitFile),
+                                         ( exists_file(CommitFile) -> true ; throw(error(no_valid_snapshot, _)) ),
+                                         %Load and deserialize continuation:
+                                         open(ContFile, read, In1),
+                                         read_term(In1, SCont, []),
+                                         close(In1),
+                                         deserialize(SCont, Cont),
+                                         %Retrieve continuation variable:
+                                         open(VarFile, read, In2),
+                                         read_term(In2, Index, []),
+                                         close(In2),
+                                         term_variables(Cont, Vars),
+                                         nth0(Index, Vars, OutVar),
+                                         %Resume continuation:
+                                         call(Resume, Cont, OutVar, X).
+
+%^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^%

run_with_cpu_limit.sh

@@ -0,0 +1,17 @@
+#!/bin/sh
+# cpu_limit.sh
+# usage: ./cpu_limit.sh <cpu_seconds> command [args...]
+
+if [ $# -lt 2 ]; then
+  echo "Usage: $0 <cpu_seconds> command [args...]"
+  exit 1
+fi
+
+CPU_LIMIT="$1"
+shift
+
+# set CPU time limit (in seconds)
+ulimit -t "$CPU_LIMIT"
+
+# run command (replaces shell with it)
+exec "$@"

test.sh

@@ -25,7 +25,7 @@ pidfile="/tmp/metta_pid_map.$$"
 
 for f in ./examples/*.metta; do
     base=$(basename "$f")
-    case "$base" in repl.metta|llm_cities.metta|torch.metta|greedy_chess.metta|git_import2.metta)
+    case "$base" in snapshot_example.metta|repl.metta|llm_cities.metta|torch.metta|greedy_chess.metta|git_import2.metta)
         continue ;;
     esac
     run_test "$f" &