state_exec: the stdlib materializers run as sliced data steps, and predicates get them

bindings/pycvc/test_pycvc_exec.py

@@ -335,6 +335,54 @@ def run_error(src):
     assert ex.run("(length (collections.range 1000))") == "1000"
 
 
+def test_stdlib_collections_take_dsl_and_python_functions():
+    # map/filter/reduce call a lambda as the program's own steps, and a registered Python
+    # function like any native one. The list comes first, then the function.
+    app = pycvc.make_app()
+    ex = pycvc.Exec(app)
+    assert ex.run("(collections.map (list 1 2 3) (lambda (x) (* x x)))") == "(1 4 9)"
+    assert ex.run("(collections.filter (collections.range 10) (lambda (x) (= (% x 3) 0)))") == (
+        "(0 3 6 9)"
+    )
+    assert ex.run("(collections.reduce (list 1 2 3 4) (lambda (acc x) (+ acc (* x x))) 0)") == "30"
+    ex.register_fn("twice", lambda x: x * 2)
+    assert ex.run("(collections.map (list 1 2) twice)") == "(2 4)"
+    # The builtins that call a function, apply and send, take a DSL function too.
+    assert ex.run("(apply (lambda (a b) (* a b)) (list 6 7))") == "42"
+    assert ex.run("(apply twice (list 21))") == "42"
+    assert (
+        ex.run(
+            '(begin (set o (dict "__class__" (dict "inc" (lambda (self x) (+ x 1)))))'
+            ' (send o "inc" 41))'
+        )
+        == "42"
+    )
+
+
+def test_stdlib_materializers_run_as_data_steps():
+    # Exec slices deep-data work 1024 nodes a step (the host default): a big join and split
+    # spread over data steps, and the answer is the same however the work is sliced.
+    app = pycvc.make_app()
+    ex = pycvc.Exec(app)
+    prog = (
+        '(begin (set parts (string.split (string.join (collections.map (collections.range 20000)'
+        ' str) ",") ","))'
+        " (list (length parts) (nth parts 12345) (get-attr (budget-usage) \"data_steps\")))"
+    )
+
+    def data_steps(step_nodes=None):
+        if step_nodes is not None:
+            pycvc.state_set(app, "state_exec.defaults.data.step_nodes", str(step_nodes))
+        out = ex.run(prog)
+        assert out.startswith('(20000 "12345" '), out
+        return int(out.rsplit(" ", 1)[1].rstrip(")"))
+
+    host = data_steps()
+    assert host > 100  # map, join and split, a slice a step
+    assert data_steps(16) > host
+    assert data_steps(0) < 10  # whole, in the steps that need it
+
+
 # ── reference cycles ────────────────────────────────────────────────────
 
 

docs/STATE_EXEC_DEVELOPER_GUIDE.md

@@ -134,6 +134,16 @@ The full syntax (literals, escapes, Unicode rules) is specified in
 **OOP:** `send` (method dispatch), `apply`
 **Stdlib:** `string.*`, `bytes.*`, `math.*`, `collections.*`, always loaded ([§11](#11-standard-library))
 
+`apply`, `send` and `collections.map`/`filter`/`reduce` call a native function or a DSL
+function alike. A DSL function runs as the program's own frames on the same evaluator
+(`call_function()` in `data_job.h`; see §11, "Sliced work"):
+
+```lisp
+(apply (lambda (a b) (* a b)) (list 6 7))        ;; => 42
+(set o (dict "__class__" (dict "inc" (lambda (self x) (+ x 1)))))
+(send o "inc" 41)                                ;; => 42
+```
+
 #### Type Conversion
 
 ```lisp
@@ -1399,11 +1409,11 @@ mine.import_module("geo", env, {"geo.area"});         // or a subset
 
 ### Allocation limit
 
-A native call builds its whole result inside one evaluator step. The per-process
-`max_memory` check runs between steps, and the heap it counts is measured after the fact (§7), so
-neither can stop `(collections.range 1000000000)` from allocating. Each function whose result can be larger than all of its
-arguments therefore estimates the result's size and fails before allocating when it is over
-the limit:
+The per-process `max_memory` check runs between steps, and the heap it counts is measured after
+the fact (§7), so neither can stop `(collections.range 1000000000)` from allocating. Each
+function whose result can be larger than all of its arguments therefore works out the result's
+size and fails before building it when it is over the limit. A sliced function (next section)
+checks as soon as it has counted, before it builds anything. The checked functions are:
 
 - `string.split`, `string.join`, `string.replace`, `bytes.concat`;
 - `collections.range`, `collections.zip`, `collections.flatten`, `collections.dict-merge`;
@@ -1433,24 +1443,56 @@ process with it as the exit error; the sync scheduler's `run()` throws it:
 ;;        16777216 bytes (limits.max_alloc_bytes, or what is left of max_memory)
 ```
 
-`collections.unique` and `collections.dict-merge` are quadratic, so they also poll the
-evaluation deadline (`max_time`) as they go.
+### Sliced work
+
+These functions do their work as a data job (`data_job.h`; "Deep data on the executor" in
+`docs/STATE_EXEC_GRAMMAR.md`): `string.split`, `string.join`, `string.replace`,
+`string.encode`, `bytes.slice`, `bytes.concat`, `bytes.decode` and every `collections.*`
+function.
+
+- **Slices.** A slice is `data.step_nodes` nodes: one per element visited, and one per 256
+  bytes (`kDataNodeBytes`) scanned or copied. With `data.step_nodes` = N the call does one
+  slice in its own step. The rest runs N nodes per **data step** while the call waits. Data
+  steps count against `max_steps` and `max_time`, and other processes run between them. A call
+  that fits in one slice takes no extra step.
+- **Defaults.** The app's exec scheduler, pycvc's `Exec` and Ariadne predicates slice 1024
+  nodes a step (`async_scheduler::host_data_defaults()`). With 0, the sync scheduler's default,
+  the work runs whole in the call's step.
+- **Calls.** `collections.map`, `filter` and `reduce` hand each call of their function to the
+  evaluator running the program, and so do the builtins `apply` and `send`. A native function
+  is called inside the data step. A DSL function runs as the process's own frames, so its
+  body's steps are eval steps. The job goes on in the data step after it returns, even with
+  `data.step_nodes` = 0. Another native function can do the same with
+  `call_function(fn, args, who)`.
+- **Cost.** `collections.sort` is a stable merge sort, n log n nodes.
+  `collections.unique` costs a node per comparison, and compares lists and dicts with a sliced
+  equality walk. `collections.dict-merge` keeps a key index, so it is linear. Every job polls the
+  evaluation deadline (`max_time`) about every 64 nodes.
+- **While a call runs.** Signals and watch events wait until the call is done, and `fork`
+  returns -1. A snapshot taken during it restores to the call's start: the call runs again, and
+  calls its function again from the first element.
+- **Measured** (`state_exec_bench_test` `StdlibJobThroughput`, 1024-node steps, MSVC Release):
+  - about 5-20 ns a node for split, join, replace and sort;
+  - about 200-370 ns a node for a 256-byte copy such as `string.encode`, including first-touch
+    page faults;
+  - about 60 ns an element for `collections.map` with a native function, and about 400 ns with
+    `(lambda (x) x)`, whose body's eval steps are included.
 
 ### In Ariadne predicates
 
 A per-frame predicate (`visible_when`, `bind`, `fmt`, …) runs in a reduced environment. It
-gets the stdlib functions whose result is a scalar or no bigger than their argument:
+gets every stdlib function except `math.random`, which would make a predicate
+nondeterministic.
 
-- `string.length`, `string.byte-length`, `string.char-at`, `string.substring`,
-  `string.upper`, `string.lower`, `string.trim`, `string.starts-with`, `string.ends-with`,
-  `string.contains`;
-- `bytes.length`, `bytes.byte-at`;
-- every `math.*` function except `math.random`, which would make a predicate
-  nondeterministic.
+The materializers are safe there because their work is sliced. The step and time caps are
+checked between steps, and a predicate always slices: it uses the app tree's
+`state_exec.defaults.data.step_nodes`, with 1024 for 0 or unset. A huge call is cut by the caps
+(fail-safe, with a warning) instead of stalling the frame, and each call stays under the 16 MiB
+allocation limit.
 
-The rest (`string.join`/`replace`/`split`/`encode`, `bytes.slice`/`concat`/`decode` and all of
-`collections.*`) are unbound in a predicate until they run as scheduled work, a slice per
-step. The `init:` and action lanes get every function.
+A function a predicate passes to `collections.map`, `filter` or `reduce` runs on the
+predicate's own evaluator. The special-form gate, the reduced environment and the caps cover
+its body too. The `init:` and action lanes get every function.
 
 ### Module: `string` (14 functions)
 
@@ -1510,17 +1552,22 @@ text and bytes.
 
 ### Module: `collections` (12 functions)
 
-`collections.map`, `filter` and `reduce` take the list first, then a native function: a
-builtin or a stdlib function. A DSL `lambda` is not accepted yet; it needs the scheduled-work
-mechanism above.
+`collections.map`, `filter` and `reduce` take the list first, then a function. It can be a
+builtin, a stdlib function, a host function (pycvc's `register_fn`) or a DSL function
+(`lambda`, `defun`). A DSL function runs as the program's own steps (see Sliced work), and
+`return` in it returns from it. C++ that calls the registry's function directly, with no
+running program, can pass only a native function.
 
 ```lisp
 (collections.map (list 1 4 9) math.sqrt)          ;; => (1 2 3)
+(collections.map (list 1 2 3) (lambda (x) (* x x))) ;; => (1 4 9)
 (collections.filter (list 1 "a" 2) is-int)        ;; => (1 2)
+(collections.filter (list 1 2 3 4) (lambda (x) (= (% x 2) 0))) ;; => (2 4)
 (collections.reduce (list 1 2 3) + 0)             ;; => 6
+(collections.reduce (list 1 2 3) (lambda (acc x) (+ acc (* x x))) 0) ;; => 14
 (collections.zip (list 1 2) (list "a" "b"))       ;; => ((1 "a") (2 "b"))
 (collections.flatten (list (list 1 2) 3 (list 4))) ;; => (1 2 3 4)   one level
-(collections.sort (list 3 1 2))                   ;; => (1 2 3)
+(collections.sort (list 3 1 2))                   ;; => (1 2 3)   numbers by value, strings by bytes; stable
 (collections.reverse (list 1 2 3))                ;; => (3 2 1)
 (collections.range 1 5)                           ;; => (1 2 3 4)
 (collections.range 5 0 -2)                        ;; => (5 3 1)

docs/STATE_EXEC_GRAMMAR.md

@@ -240,17 +240,26 @@ A node is an element visited: a list element or a dict entry walked, or an eleme
   other processes run between them.
 - **Signals and watch events** wait while a walk is pending, because the walk's call frame is
   on top of the evaluation stack.
+- **Function calls.** A job can stop for a call of a function (`collections.map`, `filter`,
+  `reduce`, and `apply`/`send` through `call_function()`), and the evaluator makes it. A native function is called inside the data step. A
+  DSL function runs as frames above the waiting call, in eval steps; `return` and `break` in it
+  unwind no further than that call. The job goes on in the data step after the function
+  returns. Such a job waits on the evaluator even with `step_nodes` = 0.
 - **Snapshots** keep `step_nodes` but not a walk in flight: after a restore, the waiting call
-  or macro expansion runs again.
+  or macro expansion runs again. The frames of a DSL function it was calling are left out.
 - **Values changed between slices.** A walk holds the lists and dicts it walks by owning
   pointer and reads them by index, checking bounds at every node. A change between two slices
   (by another process) changes the answer, never memory safety.
 
-Other deep walks run whole inside one step, as before, but count their nodes. These are
-`collections.*` helpers and string coercion for `state-set`.
+The stdlib's materializers are jobs too: `string.split`/`join`/`replace`/`encode`,
+`bytes.slice`/`concat`/`decode` and all of `collections.*`. A string function counts a node per
+256 bytes it scans or copies (developer guide §11, "Sliced work"). String coercion for
+`state-set` still runs whole inside one step, but counts its nodes.
 
 Measured per node (MSVC Release, `state_exec_bench_test` `DataWalkThroughput`): ~45 ns to free,
 ~70 ns to print, ~250 ns to compare and ~380 ns to copy. So a 1024-node slice costs 0.05-0.4 ms.
+The stdlib jobs (`StdlibJobThroughput`) cost 5-20 ns a node for elements and 200-370 ns for a
+256-byte copy.
 
 ### Budget by kind
 

inc/cvc/state/state_exec/data_job.h

@@ -15,8 +15,11 @@
 #include <cvc/state/state_exec/types.h>
 #include <functional>
 #include <memory>
+#include <optional>
+#include <span>
 #include <string>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 namespace cvc {
@@ -27,6 +30,10 @@ namespace cvc::state_exec {
 
 struct evaluation_stats;
 
+/// Byte work (scanning, copying) counts one node per this many bytes, so a slice of a string
+/// function costs about what a slice of a walk does.
+inline constexpr std::size_t kDataNodeBytes = 256;
+
 /// A walk over a nested value, done a slice at a time. A job holds what it walks (the shared
 /// lists and dicts, by owning pointer) and reads them by index, checking bounds at every node,
 /// so a value changed between two slices (by another process) cannot leave it dangling; the
@@ -36,6 +43,9 @@ class data_job {
   virtual ~data_job() = default;
 
   /// Visit at most `nodes` more nodes; 0 visits all that remain. True once the walk is done.
+  /// A job that applies a function (collections.map and friends) also stops when it needs a
+  /// call made: step() returns false with a call pending (take_call()). Whoever runs the job
+  /// makes the call, passes its value to call_returned() and steps the job again.
   virtual bool step(std::size_t nodes) = 0;
 
   /// The walk's result, once step() has returned true. Call it once.
@@ -44,8 +54,28 @@ class data_job {
   /// Nodes visited so far.
   std::size_t nodes() const { return nodes_; }
 
+  /// A function application a job waits for. `fn` and `args` belong to the job and stay valid
+  /// until it is stepped again.
+  struct call_request {
+    const value_t *fn;
+    std::span<const value_t> args;
+    const char *who; // the stdlib function making the call, for errors
+  };
+
+  /// The call the job stopped for, if any (taking it clears it).
+  std::optional<call_request> take_call() { return std::exchange(call_, std::nullopt); }
+  const call_request *pending_call() const { return call_ ? &*call_ : nullptr; }
+
+  /// The value of the call the job stopped for.
+  virtual void call_returned(value_t value) { (void)value; }
+
+  /// Called once the job outlives the native call that made it: copy what it only borrowed
+  /// from that call's arguments.
+  virtual void own_inputs() {}
+
 protected:
   std::size_t nodes_ = 0;
+  std::optional<call_request> call_;
 };
 
 /// Whether walking `v` visits any nodes: it is a non-empty list or dict.
@@ -73,7 +103,9 @@ std::unique_ptr<data_job> make_copy_job(value_t v);
 std::unique_ptr<data_job> make_substitute_job(value_t tmpl,
                                               std::unordered_map<std::string, value_t> subst);
 
-/// Run a job to completion and return its result.
+/// Run a job to completion and return its result. The calls it makes are made here: a native
+/// function's own deep-data work then runs here too, and a DSL function (a closure) throws,
+/// since only an evaluator can run one.
 value_t run_to_completion(data_job &job);
 
 /// Deep-data settings ("Deep data on the executor" in docs/STATE_EXEC_GRAMMAR.md). Hosts keep them
@@ -93,15 +125,24 @@ struct data_options {
 data_options data_options_from_state(cvc::state &root, const std::string &scheduler_id = {},
                                      const data_options &base = {});
 
-/// Do a native function's deep-data work. Inside a stackless evaluator step that slices
-/// (data.step_nodes > 0), called from the native function that step is applying, `job` does one
-/// slice here; if that finishes it, `then(result)` is returned. If not, the job and `then` pass
-/// to the evaluator, which runs a slice per step and completes the call with then's value once
-/// the job is done; what this returns is then ignored, so the native function should return it
-/// at once. Anywhere else the job runs to completion here. Without `then`, the job's result is
-/// the call's.
+/// Do a native function's deep-data work. Inside a stackless evaluator step, called from the
+/// native function that step is applying, `job` does one slice here (all of it when the step
+/// does not slice, data.step_nodes = 0); if that finishes it, `then(result)` is returned. If not
+/// (the slice ran out, or the job needs a function called), the job and `then` pass to the
+/// evaluator: it runs a slice per step, makes the job's calls (a DSL function as ordinary
+/// frames of the same process) and completes the call with then's value once the job is done;
+/// what this returns is then ignored, so the native function should return it at once.
+/// Anywhere else the job runs to completion here (see run_to_completion). Without `then`, the
+/// job's result is the call's.
 value_t run_data_job(std::unique_ptr<data_job> job, std::function<value_t(value_t)> then = {});
 
+/// Apply `fn`, a native function or a DSL closure, to `args` from inside a native function.
+/// A native function is called here. A closure runs through the evaluator, as a one-call job
+/// (run_data_job): inside a stackless evaluator step its body becomes the process's own frames,
+/// and the native function should return what this returns at once; anywhere else a closure
+/// throws. `who` names the caller in errors.
+value_t call_function(const value_t &fn, std::span<const value_t> args, const char *who);
+
 /// Count `nodes` units of deep-data work done in the current evaluator step (a walk run to
 /// completion counts itself; the release backlog counts what it frees).
 void count_data_nodes(std::size_t nodes);
@@ -112,6 +153,9 @@ namespace detail {
 struct deferred_data_job {
   std::unique_ptr<data_job> job;
   std::function<value_t(value_t)> then;
+  /// Set when the job waits on work a native function it called left: that runs first, and
+  /// its result is the call's value.
+  std::unique_ptr<deferred_data_job> waits_on;
 };
 
 /// The data side of one stackless evaluator step, armed for the step's duration (nested evaluator
@@ -133,6 +177,7 @@ class data_step_scope {
 
   /// Let run_data_job() defer a job while `allowed` (around the native call being applied).
   void allow_defer(bool allowed) { may_defer_ = allowed; }
+  bool defer_allowed() const { return may_defer_; }
   /// The job a native call deferred, if any.
   std::unique_ptr<deferred_data_job> take_deferred() { return std::move(deferred_); }