diff --git a/bindings/pycvc/test_pycvc_exec.py b/bindings/pycvc/test_pycvc_exec.py index d624657f3..e219f7946 100644 --- a/bindings/pycvc/test_pycvc_exec.py +++ b/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 ──────────────────────────────────────────────────── diff --git a/docs/STATE_EXEC_DEVELOPER_GUIDE.md b/docs/STATE_EXEC_DEVELOPER_GUIDE.md index 296c3563d..7bc6e359f 100644 --- a/docs/STATE_EXEC_DEVELOPER_GUIDE.md +++ b/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) diff --git a/docs/STATE_EXEC_GRAMMAR.md b/docs/STATE_EXEC_GRAMMAR.md index 00df8b858..2485bcc02 100644 --- a/docs/STATE_EXEC_GRAMMAR.md +++ b/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 diff --git a/inc/cvc/state/state_exec/data_job.h b/inc/cvc/state/state_exec/data_job.h index 52d41a141..9bdce9132 100644 --- a/inc/cvc/state/state_exec/data_job.h +++ b/inc/cvc/state/state_exec/data_job.h @@ -15,8 +15,11 @@ #include #include #include +#include +#include #include #include +#include #include 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 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 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_; }; /// Whether walking `v` visits any nodes: it is a non-empty list or dict. @@ -73,7 +103,9 @@ std::unique_ptr make_copy_job(value_t v); std::unique_ptr make_substitute_job(value_t tmpl, std::unordered_map 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 job, std::function 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 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 job; std::function 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 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 take_deferred() { return std::move(deferred_); } diff --git a/inc/cvc/state/state_exec/stackless_evaluator.h b/inc/cvc/state/state_exec/stackless_evaluator.h index b1adca36b..1173ea52c 100644 --- a/inc/cvc/state/state_exec/stackless_evaluator.h +++ b/inc/cvc/state/state_exec/stackless_evaluator.h @@ -65,16 +65,30 @@ struct eval_frame { environment_ptr extra_env; }; -/// Deep-data work a step left unfinished (data_options::step_nodes): a native call's job (its -/// frame waits in the apply phase), the expansion of a macro call (its frame waits to be -/// re-initialized with it), or a measurement of the process's heap that its scheduler started -/// between two steps (heap.h; its result goes to evaluator_state::heap_measured, and evaluation -/// resumes where it was). +/// Deep-data work a step left unfinished (data_options::step_nodes, or a job waiting on a call): +/// a native call's job (its frame waits in the apply phase), the expansion of a macro call (its +/// frame waits to be re-initialized with it), or a measurement of the process's heap that its +/// scheduler started between two steps (heap.h; its result goes to +/// evaluator_state::heap_measured, and evaluation resumes where it was). struct pending_data_work { enum class kind { native_call, macro_expansion, heap_measure }; std::unique_ptr job; std::function then; // native_call: turns the job's result into the call's kind what = kind::native_call; + /// The stack size when the work began. The frame waiting on it is stack[base - 1]; a DSL + /// function the job calls runs in the frames above it. + std::size_t base = 0; + /// How many values the waiting frame held then: the call's arguments, which a restore runs + /// the call with again. + std::size_t frame_args = 0; + /// Set while a DSL function the job called runs; its value arrives on the waiting frame, and + /// `return`/`break` in it unwind no further than that frame. + bool calling = false; + /// The work this work runs inside of, if any: the job of a call that was waiting on a DSL + /// function when this work began, or (feeds_outer) the job whose native call left this work. + std::shared_ptr outer; + /// The result is the value of a call outer's job made, not the waiting frame's. + bool feeds_outer = false; }; /// Serializable evaluator state — the entire execution snapshot. @@ -94,8 +108,9 @@ struct evaluator_state { /// it). 0 does all of it in the step that needs it. std::size_t data_step_nodes = 0; /// Set while deep-data work runs a slice per step; step() runs it instead of evaluating until - /// it is done, then completes what started it. Not serialized: after a restore the waiting - /// call or macro expansion simply runs again. + /// it is done (except while a DSL function it called runs), then completes what started it. + /// The innermost work; pending_data_work::outer chains the rest. Not serialized: after a + /// restore the call or macro expansion waiting on the outermost work simply runs again. std::shared_ptr data_work; /// Last owners this state's steps dropped and have not freed yet (data_step_nodes > 0); each /// step frees them with what is left of its slice, and steps that only free them follow @@ -210,6 +225,8 @@ class stackless_evaluator { // Helpers void call_closure(evaluator_state &state, eval_frame &frame, const closure_ptr &cls, const std::vector &args); + // Push the frame that evaluates `cls`'s body in `local` (its parameters already bound). + void push_closure_body(evaluator_state &state, const closure_ptr &cls, environment_ptr local); void handle_defclass(evaluator_state &state, eval_frame &frame, const std::vector &args); environment_ptr global_env_; diff --git a/src/cvc/ariadne/ariadne.cpp b/src/cvc/ariadne/ariadne.cpp index ef99a69cf..af6ff5c67 100644 --- a/src/cvc/ariadne/ariadne.cpp +++ b/src/cvc/ariadne/ariadne.cpp @@ -424,15 +424,21 @@ class ReactiveEngine { // side-effect-free state readers (state-data-get deep-copies; no aliasing, no callables) "state-get", "state-exists", "state-children", "state-data-get", "state-root-path", "state-has-expiry", "state-is-expired", - // the pure, bounded stdlib: a scalar, or a result no bigger than its argument. The - // materializers (string.join/replace/split/encode, bytes.slice/concat/decode, - // collections.*) stay out until they run as scheduled, sliced work; math.random stays - // out because a predicate must be deterministic. + // the stdlib: the pure, bounded functions (a scalar, or a result no bigger than its + // argument), and the materializers, which do their work as sliced data steps (each + // eval slices, see data_step_nodes_) under the per-call allocation limit. A function + // collections.map/filter/reduce call runs as this evaluator's own frames, so the + // special-form gate and the caps cover it too. math.random stays out because a + // predicate must be deterministic. "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", "math.sqrt", "math.abs", "math.floor", "math.ceil", - "math.round", "math.pow", "math.log", "math.sin", "math.cos", "math.tan", "math.min", - "math.max", "math.clamp", "math.pi", "math.e"}; + "string.split", "string.join", "string.replace", "string.encode", "bytes.length", + "bytes.byte-at", "bytes.slice", "bytes.concat", "bytes.decode", "math.sqrt", "math.abs", + "math.floor", "math.ceil", "math.round", "math.pow", "math.log", "math.sin", "math.cos", + "math.tan", "math.min", "math.max", "math.clamp", "math.pi", "math.e", "collections.map", + "collections.filter", "collections.reduce", "collections.zip", "collections.flatten", + "collections.sort", "collections.reverse", "collections.range", "collections.unique", + "collections.dict-keys", "collections.dict-values", "collections.dict-merge"}; for (const char *name : kAllowed) if (const se::value_t *v = full->lookup(name)) env_->set(name, *v); @@ -445,13 +451,17 @@ class ReactiveEngine { std::set{"if", "begin", "let", "while", "for", "lambda", "defun", "set", "return", "quote", "yield", "break"}); ev_->restrict_special_forms(kAllowedForms); + resolve_data_settings(); } // Reset the per-FRAME aggregate budget — call once at the top of each render frame, // before the walk. Once the frame's cumulative predicate-eval time is spent, the // remaining predicates this frame skip evaluation and degrade fail-safe (hidden) with a // one-time warning, so N reactive widgets cannot together stall the frame. - void begin_frame() { frame_ = se::time_budget(kFrameBudgetSeconds); } + void begin_frame() { + frame_ = se::time_budget(kFrameBudgetSeconds); + resolve_data_settings(); + } // §12: re-chroot the read-lane to `path` so a mounted fragment's predicates read state // relative to its own sub-prefix, exactly as its widget binds resolve there. The evaluator, @@ -570,6 +580,7 @@ class ReactiveEngine { }; try { se::evaluator_state st = ev_->create_state(*expr); + st.data_step_nodes = data_step_nodes_; frame_.start(); const charge_frame charge{frame_, st}; se::value_t r = ev_->run(st, kMaxSteps, std::min(kMaxSeconds, remaining)); @@ -614,11 +625,22 @@ class ReactiveEngine { } } + // The deep-data slice (data.step_nodes) every eval runs with: the tree's state_exec.defaults + // over the host default. The caps are checked only between steps, so a predicate always + // slices: a 0 there (no slicing) gets the host default here. + void resolve_data_settings() { + const se::data_options host = se::async_scheduler::host_data_defaults(); + data_step_nodes_ = se::data_options_from_state(*tree_, {}, host).step_nodes; + if (data_step_nodes_ == 0) + data_step_nodes_ = host.step_nodes; + } + static constexpr uint64_t kMaxSteps = 100'000; // per-eval step cap (a predicate is tiny) static constexpr double kMaxSeconds = 0.005; // per-eval time cap (5 ms) static constexpr double kFrameBudgetSeconds = 0.010; // aggregate across all predicates/frame static constexpr std::size_t kMaxOptions = 1024; // cap on computed combo options + std::size_t data_step_nodes_ = 0; // see resolve_data_settings() se::time_budget frame_{kFrameBudgetSeconds}; // the current frame's predicate evals' time std::string root_path_; // §12: the scope the ctx_ is currently chrooted to (set_root_path) cvc::state *tree_ = nullptr; // the app's state tree, for state_exec.defaults.parser.* diff --git a/src/cvc/state/state_exec/async_scheduler.cpp b/src/cvc/state/state_exec/async_scheduler.cpp index 7e9b2c625..6a3d91d29 100644 --- a/src/cvc/state/state_exec/async_scheduler.cpp +++ b/src/cvc/state/state_exec/async_scheduler.cpp @@ -733,6 +733,12 @@ int async_scheduler::fork(int pid) { auto &parent = *it->second; if (parent.status != process_status::ready && parent.status != process_status::running) return -1; + // Deep-data work in flight (a function collections.map calls, say) is not copied: a child + // could not finish the call its stack waits on. A heap measurement waits on no frame, so a + // child without it is whole. + if (parent.state.data_work && + parent.state.data_work->what != pending_data_work::kind::heap_measure) + return -1; int child_pid = next_pid_++; auto child = make_process(); diff --git a/src/cvc/state/state_exec/builtins.cpp b/src/cvc/state/state_exec/builtins.cpp index a55bce6cd..a28f50a3f 100644 --- a/src/cvc/state/state_exec/builtins.cpp +++ b/src/cvc/state/state_exec/builtins.cpp @@ -127,18 +127,15 @@ value_t dispatch_method(const value_t &obj, const std::string &method, while (cls_dict) { if (value_t *m = dict_find(*cls_dict, method)) { - // Invoke: first arg is self - if (auto *fn = std::get_if(&m->v)) { - std::vector call_args; - call_args.reserve(1 + args.size()); - call_args.push_back(obj); - call_args.insert(call_args.end(), args.begin(), args.end()); - return (*fn)(call_args); - } - if (auto *c = std::get_if(&m->v)) - throw std::runtime_error("send: closure method dispatch requires evaluator " - "(use evaluator-level send)"); - throw std::runtime_error("send: method is not callable"); + if (!std::holds_alternative(m->v) && !std::holds_alternative(m->v)) + throw std::runtime_error("send: method is not callable"); + // Invoke: first arg is self. A closure runs as the program's own frames. + std::vector call_args; + call_args.reserve(1 + args.size()); + call_args.push_back(obj); + call_args.insert(call_args.end(), args.begin(), args.end()); + const value_t fn = *m; // the dict may change while a closure runs + return call_function(fn, call_args, "send"); } // walk __super__ value_t *sp = dict_find(*cls_dict, "__super__"); @@ -630,12 +627,9 @@ value_t builtin_send(std::span args) { value_t builtin_apply(std::span args) { expect_exact(args, 2, "apply"); - auto *fn = std::get_if(&args[0].v); - if (!fn) - throw std::runtime_error("apply: first argument must be callable (native_fn); " - "closure dispatch requires evaluator"); auto &arg_list = as_list(args[1], "apply"); - return (*fn)(arg_list); + // A native function is called here; a closure runs as the program's own frames. + return call_function(args[0], arg_list, "apply"); } // --------------------------------------------------------------------------- diff --git a/src/cvc/state/state_exec/data_job.cpp b/src/cvc/state/state_exec/data_job.cpp index 2a9dc882e..14a7565cc 100644 --- a/src/cvc/state/state_exec/data_job.cpp +++ b/src/cvc/state/state_exec/data_job.cpp @@ -476,6 +476,39 @@ class substitute_walk final : public data_job { std::vector> open_; }; +// -- call ---------------------------------------------------------------------- + +// One application of a function, as a job, so a native function can have a closure called +// through the evaluator (call_function). +class call_walk final : public data_job { +public: + call_walk(value_t fn, std::vector args, const char *who) + : fn_(std::move(fn)), args_(std::move(args)), who_(who) {} + + bool step(std::size_t) override { + if (!asked_) { + asked_ = true; + ++nodes_; + call_ = call_request{&fn_, args_, who_}; + } + return returned_; + } + + void call_returned(value_t value) override { + result_ = std::move(value); + returned_ = true; + } + + value_t result() override { return std::move(result_); } + +private: + value_t fn_; + std::vector args_; + const char *who_; + bool asked_ = false, returned_ = false; + value_t result_; +}; + // The data side of the evaluator step running on this thread, if any. thread_local detail::data_step_scope *t_scope = nullptr; @@ -544,11 +577,44 @@ std::unique_ptr make_substitute_job(value_t tmpl, return std::make_unique(std::move(tmpl), std::move(subst)); } +namespace { +// Make a call a job running here (not on an evaluator) waits for. A native function's own +// deep-data work must run here too, or its result would be lost: deferral is off for the call. +value_t call_here(const data_job::call_request &call) { + if (const auto *fn = std::get_if(&call.fn->v)) { + struct no_defer { + detail::data_step_scope *scope = detail::data_step_scope::current(); + bool was = scope && scope->defer_allowed(); + no_defer() { + if (was) + scope->allow_defer(false); + } + ~no_defer() { + if (was) + scope->allow_defer(true); + } + } guard; + return (*fn)(call.args); + } + if (std::holds_alternative(call.fn->v)) + throw std::runtime_error(std::string(call.who) + + ": calling a DSL function needs a running program (an evaluator)"); + throw std::runtime_error(std::string(call.who) + ": not a function: " + call.fn->type_name()); +} +} // namespace + value_t run_to_completion(data_job &job) { - const std::size_t before = job.nodes(); - job.step(0); - count_data_nodes(job.nodes() - before); - return job.result(); + for (;;) { + const std::size_t before = job.nodes(); + const bool done = job.step(0); + count_data_nodes(job.nodes() - before); + if (done) + return job.result(); + std::optional call = job.take_call(); + if (!call) + throw std::logic_error("data_job: stopped with neither a result nor a call"); + job.call_returned(call_here(*call)); + } } // -- settings ------------------------------------------------------------------ @@ -573,17 +639,48 @@ data_options data_options_from_state(cvc::state &root, const std::string &schedu value_t run_data_job(std::unique_ptr job, std::function then) { detail::data_step_scope *scope = t_scope; - const bool may_defer = scope && scope->step_nodes_ > 0 && scope->may_defer_ && !scope->deferred_; - const std::size_t before = job->nodes(); - const bool done = job->step(may_defer ? scope->remaining() : 0); - count_data_nodes(job->nodes() - before); - if (!done) { - scope->deferred_ = std::make_unique( - detail::deferred_data_job{std::move(job), std::move(then)}); - return nil_value; + if (!scope || !scope->may_defer_ || scope->deferred_) { + value_t r = run_to_completion(*job); + return then ? then(std::move(r)) : r; } - value_t r = job->result(); - return then ? then(std::move(r)) : r; + // One slice here (without slicing, all of it), making the native calls the job needs on the + // way. It waits for the evaluator when the slice runs out, for a DSL function, and for the work + // a native function it called left of its own. + const bool slicing = scope->step_nodes_ > 0; + std::unique_ptr waits_on; + for (;;) { + const std::size_t before = job->nodes(); + const bool done = job->step(slicing ? scope->remaining() : 0); + count_data_nodes(job->nodes() - before); + if (done) { + value_t r = job->result(); + return then ? then(std::move(r)) : r; + } + const data_job::call_request *pending = job->pending_call(); + const native_fn *fn = pending ? std::get_if(&pending->fn->v) : nullptr; + if (!fn || (slicing && scope->nodes_ >= scope->step_nodes_)) + break; + const data_job::call_request call = *job->take_call(); + value_t value = (*fn)(call.args); + if (scope->deferred_) { + waits_on = std::move(scope->deferred_); + break; + } + job->call_returned(std::move(value)); + } + job->own_inputs(); + scope->deferred_ = std::make_unique( + detail::deferred_data_job{std::move(job), std::move(then), std::move(waits_on)}); + return nil_value; +} + +value_t call_function(const value_t &fn, std::span args, const char *who) { + if (const auto *native = std::get_if(&fn.v)) + return (*native)(args); + if (const auto *cls = std::get_if(&fn.v); cls && *cls) + return run_data_job( + std::make_unique(fn, std::vector(args.begin(), args.end()), who)); + throw std::runtime_error(std::string(who) + ": not a function: " + fn.type_name()); } void count_data_nodes(std::size_t nodes) { diff --git a/src/cvc/state/state_exec/heap.cpp b/src/cvc/state/state_exec/heap.cpp index b2b599ca7..05167745e 100644 --- a/src/cvc/state/state_exec/heap.cpp +++ b/src/cvc/state/state_exec/heap.cpp @@ -529,8 +529,14 @@ void heap_after_step(process &proc, std::size_t allocated, uint64_t state_bytes_ return; if (!proc.heap_measure_due(state_bytes_used)) return; - st.data_work = std::make_shared( - pending_data_work{make_heap_share_job(proc), {}, pending_data_work::kind::heap_measure}); + // It waits on no frame: evaluation resumes at the top of the stack, as it was, so `base` and + // `frame_args` are the whole stack and the top frame's values (a snapshot keeps all of them). + auto work = std::make_shared(); + work->job = make_heap_share_job(proc); + work->what = pending_data_work::kind::heap_measure; + work->base = st.stack.size(); + work->frame_args = st.stack.back().results.size(); + st.data_work = std::move(work); proc.heap_measuring = true; } diff --git a/src/cvc/state/state_exec/scheduler.cpp b/src/cvc/state/state_exec/scheduler.cpp index 449bb7a23..18321ec98 100644 --- a/src/cvc/state/state_exec/scheduler.cpp +++ b/src/cvc/state/state_exec/scheduler.cpp @@ -737,6 +737,12 @@ int scheduler::fork(int pid) { auto &parent = *it->second; if (parent.status != process_status::ready && parent.status != process_status::running) return -1; + // Deep-data work in flight (a function collections.map calls, say) is not copied: a child + // could not finish the call its stack waits on. A heap measurement waits on no frame, so a + // child without it is whole. + if (parent.state.data_work && + parent.state.data_work->what != pending_data_work::kind::heap_measure) + return -1; int child_pid = next_pid_++; auto child = make_process(); diff --git a/src/cvc/state/state_exec/stackless_evaluator.cpp b/src/cvc/state/state_exec/stackless_evaluator.cpp index d04a5cd37..4a0898183 100644 --- a/src/cvc/state/state_exec/stackless_evaluator.cpp +++ b/src/cvc/state/state_exec/stackless_evaluator.cpp @@ -105,21 +105,124 @@ struct native_call_scope { native_call_scope &operator=(const native_call_scope &) = delete; detail::data_step_scope *data; }; + +// New innermost deep-data work, waiting on stack[base - 1]. +std::shared_ptr +make_work(std::unique_ptr job, std::function then, + pending_data_work::kind what, std::size_t base, std::size_t frame_args, + std::shared_ptr outer, bool feeds_outer) { + auto work = std::make_shared(); + work->job = std::move(job); + work->then = std::move(then); + work->what = what; + work->base = base; + work->frame_args = frame_args; + work->outer = std::move(outer); + work->feeds_outer = feeds_outer; + return work; +} + +// Make the job a native call left the innermost work: waiting on stack[base - 1] or, with +// feeds_outer, for the call outer's job made. Work the job's own pending call left +// (deferred_data_job::waits_on) goes inside it, its result being that call's value. +void await_deferred(evaluator_state &state, detail::deferred_data_job &&deferred, std::size_t base, + std::size_t frame_args, std::shared_ptr outer, + bool feeds_outer) { + for (detail::deferred_data_job *d = &deferred; d; d = d->waits_on.get()) { + state.data_work = + make_work(std::move(d->job), std::move(d->then), pending_data_work::kind::native_call, base, + frame_args, std::move(outer), feeds_outer); + outer = state.data_work; + feeds_outer = true; + } +} + +// Whether the innermost work waits on a DSL function it called that is still running. +bool calling_out(const evaluator_state &state) { + const pending_data_work *work = state.data_work.get(); + return work && work->calling && state.stack.size() > work->base; +} + +// How far down a `return` or `break` may unwind: to the frame waiting on a DSL function a job +// called, whose value that is (0 = no such call). +std::size_t unwind_floor(const evaluator_state &state) { + const pending_data_work *work = state.data_work.get(); + return work && work->calling ? work->base : 0; +} + +// A scope for a call of `cls` with `args`: its parameters bound, extending the closure's own. +environment_ptr bind_closure_args(const closure_ptr &cls, std::span args) { + auto local = environment::extend(cls->env_snapshot); + if (cls->variadic) { + if (cls->params.empty()) + throw std::runtime_error("closure: variadic but no param name"); + local->set(cls->params.back().name, make_list(std::vector(args.begin(), args.end()))); + } else { + if (args.size() != cls->params.size()) + throw std::runtime_error("closure: expected " + std::to_string(cls->params.size()) + + " args, got " + std::to_string(args.size())); + for (size_t i = 0; i < cls->params.size(); ++i) + local->set(cls->params[i].name, args[i]); + } + return local; +} } // namespace -// Run a slice of the deep-data work a step left; once it is done, finish the native call or macro -// expansion that started it. +// Run a slice of the deep-data work a step left, making the calls its job needs: a native +// function here, a DSL function as frames on the stack (the job resumes in the step after it +// returns). Once the work is done, finish the native call or macro expansion that started it. bool stackless_evaluator::step_data(evaluator_state &state) { state.stats.increment_step(step_kind::data); detail::data_step_scope *data = detail::data_step_scope::current(); + const bool slicing = data && data->step_nodes() > 0; const std::shared_ptr work = state.data_work; - const std::size_t before = work->job->nodes(); - const bool finished = work->job->step(data ? data->remaining() : 0); - count_data_nodes(work->job->nodes() - before); - if (!finished) - return false; - state.data_work.reset(); - value_t result = work->job->result(); + data_job &job = *work->job; + if (work->calling) { + // The DSL function the job called has returned: its value is the waiting frame's last. + work->calling = false; + std::vector &results = state.stack.back().results; + value_t value = std::move(results.back()); + results.pop_back(); + job.call_returned(std::move(value)); + } + for (;;) { + std::optional call = job.take_call(); // one it stopped for before + if (!call) { + const std::size_t before = job.nodes(); + const bool finished = job.step(slicing ? data->remaining() : 0); + count_data_nodes(job.nodes() - before); + if (finished) + break; + call = job.take_call(); + if (!call) + return false; // the slice is spent + } + if (auto *cp = std::get_if(&call->fn->v)) { + environment_ptr local = bind_closure_args(*cp, call->args); + work->calling = true; + push_closure_body(state, *cp, std::move(local)); + return false; + } + auto *fn = std::get_if(&call->fn->v); + if (!fn) + throw std::runtime_error(std::string(call->who) + + ": not a function: " + call->fn->type_name()); + value_t value; + { + native_call_scope scope(data); + value = (*fn)(call->args); + } + if (auto deferred = data ? data->take_deferred() : nullptr) { + // The function left deep-data work of its own: it runs first, and its result is the call's. + await_deferred(state, std::move(*deferred), work->base, work->frame_args, work, true); + return false; + } + job.call_returned(std::move(value)); + if (slicing && data->nodes() >= data->step_nodes()) + return false; + } + state.data_work = work->outer; + value_t result = job.result(); if (work->what == pending_data_work::kind::heap_measure) { // Record the measurement for the scheduler; evaluation resumes where it was. if (const auto *bytes = std::get_if(&result.v)) @@ -134,12 +237,15 @@ bool stackless_evaluator::step_data(evaluator_state &state) { native_call_scope call(data); // `then` may end in another job result = work->then(std::move(result)); if (auto deferred = data ? data->take_deferred() : nullptr) { - state.data_work = std::make_shared( - pending_data_work{std::move(deferred->job), std::move(deferred->then), - pending_data_work::kind::native_call}); + await_deferred(state, std::move(*deferred), work->base, work->frame_args, work->outer, + work->feeds_outer); return false; } } + if (work->feeds_outer) { + work->outer->job->call_returned(std::move(result)); + return false; + } pop_frame(state, std::move(result)); return state.done; } @@ -166,7 +272,7 @@ bool stackless_evaluator::step(evaluator_state &state) { bool done; if (state.reading) { done = step_read(state); - } else if (state.data_work) { + } else if (state.data_work && !calling_out(state)) { done = step_data(state); } else if (backlog && !backlog->empty()) { // A step that only frees what earlier steps dropped. @@ -626,8 +732,8 @@ void stackless_evaluator::step_init(evaluator_state &state, eval_frame &frame) { const bool finished = job->step(data && data->step_nodes() > 0 ? data->remaining() : 0); count_data_nodes(job->nodes()); if (!finished) { - state.data_work = std::make_shared( - pending_data_work{std::move(job), {}, pending_data_work::kind::macro_expansion}); + state.data_work = make_work(std::move(job), {}, pending_data_work::kind::macro_expansion, + state.stack.size(), frame.results.size(), state.data_work, false); return; } begin_expansion(state, job->result()); @@ -687,9 +793,8 @@ void stackless_evaluator::step_apply(evaluator_state &state, eval_frame &frame) } if (auto deferred = data ? data->take_deferred() : nullptr) { // The call left deep-data work: this frame waits while step_data() runs it. - state.data_work = std::make_shared( - pending_data_work{std::move(deferred->job), std::move(deferred->then), - pending_data_work::kind::native_call}); + await_deferred(state, std::move(*deferred), state.stack.size(), frame.results.size(), + state.data_work, false); return; } pop_frame(state, std::move(result)); @@ -875,8 +980,9 @@ void stackless_evaluator::step_let_body(evaluator_state &state, eval_frame &fram void stackless_evaluator::step_return_value(evaluator_state &state, eval_frame &frame) { auto val = frame.results.back(); state.stack.pop_back(); - // Unwind to function boundary - while (!state.stack.empty()) { + // Unwind to function boundary; a DSL function a deep-data job called returns to the job. + const std::size_t floor = unwind_floor(state); + while (state.stack.size() > floor) { auto &top = state.stack.back(); if (top.phase == eval_phase::apply) { state.stack.pop_back(); @@ -910,8 +1016,10 @@ void stackless_evaluator::step_yield_value(evaluator_state &state, eval_frame &f void stackless_evaluator::step_break_unwind(evaluator_state &state, eval_frame &frame) { auto val = frame.results.back(); state.stack.pop_back(); // Remove the break frame - // Unwind to loop boundary - while (!state.stack.empty()) { + // Unwind to loop boundary, but not out of a DSL function a deep-data job called: there the + // value is the function's. + const std::size_t floor = unwind_floor(state); + while (state.stack.size() > floor) { auto &top = state.stack.back(); if (top.phase == eval_phase::while_cond || top.phase == eval_phase::while_body || top.phase == eval_phase::for_body) { @@ -1013,21 +1121,15 @@ void stackless_evaluator::step_super_args(evaluator_state &state, eval_frame &fr void stackless_evaluator::call_closure(evaluator_state &state, eval_frame & /*frame*/, const closure_ptr &cls, const std::vector &args) { - auto local = environment::extend(cls->env_snapshot); - if (cls->variadic) { - if (cls->params.empty()) - throw std::runtime_error("closure: variadic but no param name"); - local->set(cls->params.back().name, make_list(std::vector(args.begin(), args.end()))); - } else { - if (args.size() != cls->params.size()) - throw std::runtime_error("closure: expected " + std::to_string(cls->params.size()) + - " args, got " + std::to_string(args.size())); - for (size_t i = 0; i < cls->params.size(); ++i) - local->set(cls->params[i].name, args[i]); - } + environment_ptr local = bind_closure_args(cls, args); // before the pop: args are the frame's // Replace current frame with body evaluation state.stack.pop_back(); + push_closure_body(state, cls, std::move(local)); +} + +void stackless_evaluator::push_closure_body(evaluator_state &state, const closure_ptr &cls, + environment_ptr local) { if (cls->body.size() == 1) { push_frame(state, cls->body[0], local); } else { diff --git a/src/cvc/state/state_exec/state_value_codec.cpp b/src/cvc/state/state_exec/state_value_codec.cpp index cd9c0b893..97c737fe5 100644 --- a/src/cvc/state/state_exec/state_value_codec.cpp +++ b/src/cvc/state/state_exec/state_value_codec.cpp @@ -5,6 +5,7 @@ cvc::state subtrees. */ +#include #include #include #include @@ -673,11 +674,31 @@ void encode_evaluator_state(cvc::state &node, const evaluator_state &es) { // Done flag node("done").value(std::string(es.done ? "1" : "0")); - // Frame stack + // Frame stack. Deep-data work in flight is not kept (see data_step_nodes below), and the frame + // waiting on the outermost work runs its call again after a restore: the frames of DSL + // functions that work called are left out, and the frame keeps only its arguments. { + // A heap measurement waits on no frame, so it never shortens the stack. + std::size_t keep = es.stack.size(); + std::size_t frame_args = 0; + bool trims = false; + for (const pending_data_work *w = es.data_work.get(); w; w = w->outer.get()) { + if (w->what == pending_data_work::kind::heap_measure) + continue; + keep = std::min(w->base, es.stack.size()); + frame_args = w->frame_args; + trims = true; + } state_list sl(node("stack")); - for (auto &f : es.stack) - encode_frame(sl.push_back(), f); + for (std::size_t i = 0; i < keep; ++i) { + if (i + 1 == keep && trims && es.stack[i].results.size() > frame_args) { + eval_frame waiting = es.stack[i]; + waiting.results.resize(frame_args); + encode_frame(sl.push_back(), waiting); + } else { + encode_frame(sl.push_back(), es.stack[i]); + } + } } // Global environment diff --git a/src/cvc/state/state_exec/stdlib.cpp b/src/cvc/state/state_exec/stdlib.cpp index 40c9c7136..62d301b65 100644 --- a/src/cvc/state/state_exec/stdlib.cpp +++ b/src/cvc/state/state_exec/stdlib.cpp @@ -1,12 +1,17 @@ #include #include #include +#include // the sliced work of the materializers #include #include // SE-1: codepoint-correct length/index/slice +#include #include #include +#include #include #include +#include +#include namespace cvc::state_exec { @@ -55,12 +60,6 @@ double as_number(const value_t &v, const char *name) { throw std::runtime_error(std::string(name) + ": expected number, got " + v.type_name()); } -const std::vector &as_list(const value_t &v, const char *name) { - if (auto *p = std::get_if(&v.v)) - return **p; - throw std::runtime_error(std::string(name) + ": expected list, got " + v.type_name()); -} - // Allocation estimates for eval_alloc_check(), saturating so a huge count never wraps small. std::size_t sat_add(std::size_t a, std::size_t b) { return a > SIZE_MAX - b ? SIZE_MAX : a + b; } @@ -76,6 +75,895 @@ void poll_deadline(const char *name) { throw std::runtime_error(std::string(name) + ": evaluation exceeded time budget"); } +// A list argument by owning pointer (a null one reads as empty). +list_ptr as_list_ptr(const value_t &v, const char *name) { + if (auto *p = std::get_if(&v.v)) + return *p ? *p : std::make_shared>(); + throw std::runtime_error(std::string(name) + ": expected list, got " + v.type_name()); +} + +dict_ptr as_dict_ptr(const value_t &v, const char *name, const char *what) { + if (auto *p = std::get_if(&v.v)) + return *p ? *p : std::make_shared(); + throw std::runtime_error(std::string(name) + ": " + what); +} + +// A function argument: a native function or a DSL closure. +const value_t &as_function(const value_t &v, const char *name, const char *which) { + if (std::holds_alternative(v.v) || + (std::holds_alternative(v.v) && std::get(v.v))) + return v; + throw std::runtime_error(std::string(name) + ": " + which + " argument must be a function"); +} + +// The order collections.sort sorts by: numbers by value, strings by bytes; anything else is +// unordered (a merge sort keeps such elements stable, whatever the comparison says). +bool sort_less(const value_t &a, const value_t &b) { + if (auto *ai = std::get_if(&a.v)) { + if (auto *bi = std::get_if(&b.v)) + return *ai < *bi; + if (auto *bd = std::get_if(&b.v)) + return static_cast(*ai) < *bd; + } + if (auto *ad = std::get_if(&a.v)) { + if (auto *bi = std::get_if(&b.v)) + return *ad < static_cast(*bi); + if (auto *bd = std::get_if(&b.v)) + return *ad < *bd; + } + if (auto *as = std::get_if(&a.v)) { + if (auto *bs = std::get_if(&b.v)) + return *as < *bs; + } + return false; +} + +// --------------------------------------------------------------------------- +// Sliced work. The functions that build a result, scan a string or call a function do it as a +// data_job (data_job.h), a slice at a time: on an evaluator that slices (data.step_nodes > 0) a +// call spreads over as many steps as it needs, each counted against max_steps and max_time, and +// the functions a job calls (collections.map/filter/reduce) run as the evaluator's own frames. +// A slice is so many nodes: one per element visited, one per kDataNodeBytes bytes scanned or +// copied. The allocation check comes first, once the result's size is known. A job reads its +// string arguments in place during the call's own slice and copies them only if it outlives the +// call (own_inputs); lists and dicts it holds by owning pointer and reads by index, so one +// changed between two slices changes the answer, never memory safety. +// --------------------------------------------------------------------------- + +constexpr std::size_t npos = std::string_view::npos; + +// One step() call's share of the work: `budget` nodes (0 = no limit), counted into the job. +class slice { +public: + slice(std::size_t &nodes, const char *who, std::size_t budget) + : nodes_(nodes), who_(who), budget_(budget) {} + + bool spent() const { return budget_ != 0 && used_ >= budget_; } + // What is left for a nested walk (0 = no limit; never 0 while limited). + std::size_t left() const { return budget_ == 0 ? 0 : budget_ > used_ ? budget_ - used_ : 1; } + // How many of the next `n` bytes it covers: all without a limit, else at least one node's worth. + std::size_t bytes(std::size_t n) const { + const std::size_t nodes = left(); + return nodes == 0 || nodes > n / kDataNodeBytes ? n : nodes * kDataNodeBytes; + } + void count(std::size_t nodes) { + const std::size_t before = nodes_; + nodes_ += nodes; + used_ += nodes; + if ((before >> 6) != (nodes_ >> 6)) + poll_deadline(who_); // about every 64 nodes + } + void count_bytes(std::size_t n) { count((n + kDataNodeBytes - 1) / kDataNodeBytes); } + +private: + std::size_t &nodes_; + const char *who_; + std::size_t budget_; + std::size_t used_ = 0; +}; + +class stdlib_job : public data_job { +protected: + explicit stdlib_job(const char *who) : who_(who) {} + + slice begin(std::size_t budget) { return slice(nodes_, who_, budget); } + + // Stop for `fn` applied to `args`, both the job's own (see data_job::call_request). + bool call(const value_t &fn, std::span args) { + call_ = call_request{&fn, args, who_}; + return false; + } + + // Copy what `view` still has from `from` into `store` and point it there (own_inputs). + static void own(std::string_view &view, std::string &store, std::size_t from = 0) { + store.assign(view.substr(std::min(from, view.size()))); + view = store; + } + + // Append `part` from `off` to `out`, a slice's worth at a time. True once all of it is in. + static bool append_part(std::string &out, std::string_view part, std::size_t &off, slice &sl) { + off = std::min(off, part.size()); // the part changed between slices + while (off < part.size()) { + if (sl.spent()) + return false; + const std::size_t n = sl.bytes(part.size() - off); + out.append(part.substr(off, n)); + off += n; + sl.count_bytes(n); + } + return true; + } + + const char *who_; +}; + +// The first match of `needle` (not empty) in `hay` that starts in [pos, pos + window). +std::size_t find_from(std::string_view hay, std::string_view needle, std::size_t pos, + std::size_t window) { + if (pos >= hay.size()) + return npos; + const std::size_t end = std::min(hay.size(), sat_add(sat_add(pos, window), needle.size() - 1)); + const std::size_t at = hay.substr(pos, end - pos).find(needle); + return at == npos ? npos : pos + at; +} + +// Counts the non-overlapping matches of `needle` (not empty) in `hay`, left to right. +struct match_counter { + std::size_t pos = 0; + std::size_t count = 0; + + bool run(std::string_view hay, std::string_view needle, slice &sl) { + while (pos < hay.size()) { + if (sl.spent()) + return false; + const std::size_t window = sl.bytes(hay.size() - pos); + const std::size_t at = find_from(hay, needle, pos, window); + if (at == npos) { + pos += window; + sl.count_bytes(window); + continue; + } + sl.count_bytes(at + needle.size() - pos); + ++count; + pos = at + needle.size(); + } + return true; + } +}; + +class split_job final : public stdlib_job { +public: + split_job(std::string_view s, std::string_view delim) + : stdlib_job("string.split"), s_(s), delim_(delim) {} + + void own_inputs() override { + own(s_, s_store_); + own(delim_, delim_store_); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (!sized_) { + // Every part costs a list element, so many short parts outgrow the string: count first. + if (!(delim_.empty() ? count_codepoints(sl) : matches_.run(s_, delim_, sl))) + return false; + sized_ = true; + const std::size_t count = delim_.empty() ? codepoints_ : matches_.count + 1; + eval_alloc_check(sat_add(list_bytes(count), s_.size()), "string.split"); + parts_.reserve(count); + } + return delim_.empty() ? split_codepoints(sl) : split_at_delim(sl); + } + + value_t result() override { return make_list(std::move(parts_)); } + +private: + bool count_codepoints(slice &sl) { + while (pos_ < s_.size()) { + if (sl.spent()) + return false; + const std::size_t start = pos_, end = pos_ + sl.bytes(s_.size() - pos_); + while (pos_ < end) { + char32_t cp; + std::size_t len; + utf8::decode(s_, pos_, cp, len); + pos_ += len; + ++codepoints_; + } + sl.count_bytes(pos_ - start); + } + pos_ = 0; + return true; + } + + // SE-1 (A3): split into CODEPOINTS, not bytes — each element is one whole UTF-8 codepoint, so + // a multibyte character is never shredded into invalid fragments. + bool split_codepoints(slice &sl) { + while (pos_ < s_.size()) { + if (sl.spent()) + return false; + char32_t cp; + std::size_t len; + utf8::decode(s_, pos_, cp, len); + parts_.emplace_back(std::string(s_.substr(pos_, len))); + pos_ += len; + sl.count(1); + } + return true; + } + + bool split_at_delim(slice &sl) { + for (;;) { + if (sl.spent()) + return false; + std::size_t at = npos; + if (scan_ < s_.size()) { + const std::size_t window = sl.bytes(s_.size() - scan_); + at = find_from(s_, delim_, scan_, window); + if (at == npos) { + scan_ += window; + sl.count_bytes(window); + if (scan_ < s_.size()) + continue; + } + } + if (at == npos) { + parts_.emplace_back(std::string(s_.substr(start_))); + sl.count(1); + return true; + } + parts_.emplace_back(std::string(s_.substr(start_, at - start_))); + sl.count_bytes(at + delim_.size() - scan_); + sl.count(1); + start_ = scan_ = at + delim_.size(); + } + } + + std::string_view s_, delim_; + std::string s_store_, delim_store_; + bool sized_ = false; + match_counter matches_; + std::size_t codepoints_ = 0; + std::size_t pos_ = 0; // codepoint splitting + std::size_t start_ = 0; // the part being found + std::size_t scan_ = 0; // where the search for its end goes on + std::vector parts_; +}; + +class join_job final : public stdlib_job { +public: + join_job(list_ptr list, std::string_view sep) + : stdlib_job("string.join"), list_(std::move(list)), sep_(sep) {} + + void own_inputs() override { own(sep_, sep_store_); } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (!sized_) { + for (; i_ < list_->size(); ++i_) { + if (sl.spent()) + return false; + total_ = sat_add(total_, as_string((*list_)[i_], "string.join").size()); + sl.count(1); + } + if (i_ > 0) + total_ = sat_add(total_, alloc_bytes(i_ - 1, sep_.size())); + eval_alloc_check(total_, "string.join"); + out_.reserve(total_); + sized_ = true; + i_ = 0; + } + while (i_ < list_->size()) { + if (sl.spent()) + return false; + if (i_ > 0 && !sep_done_) { + if (!append_part(out_, sep_, off_, sl)) + return false; + sep_done_ = true; + off_ = 0; + } + const std::string &part = as_string((*list_)[i_], "string.join"); + if (const std::size_t need = sat_add(out_.size(), part.size()); need > total_) { + eval_alloc_check(need, "string.join"); // the list grew between slices + total_ = need; + } + if (!append_part(out_, part, off_, sl)) + return false; + sl.count(1); + ++i_; + off_ = 0; + sep_done_ = false; + } + return true; + } + + value_t result() override { return value_t(std::move(out_)); } + +private: + list_ptr list_; + std::string_view sep_; + std::string sep_store_; + bool sized_ = false; + std::size_t total_ = 0; + std::size_t i_ = 0; + std::size_t off_ = 0; + bool sep_done_ = false; + std::string out_; +}; + +class replace_job final : public stdlib_job { +public: + replace_job(std::string_view s, std::string_view from, std::string_view to) + : stdlib_job("string.replace"), s_(s), from_(from), to_(to) {} + + void own_inputs() override { + own(s_, s_store_); + own(from_, from_store_); + own(to_, to_store_); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (!sized_) { + // Non-overlapping matches, left to right. A longer `to` multiplies the size: count first. + if (!matches_.run(s_, from_, sl)) + return false; + sized_ = true; + const std::size_t count = matches_.count; + if (count > 0) { + const std::size_t size = + sat_add(s_.size() - count * from_.size(), alloc_bytes(count, to_.size())); + eval_alloc_check(size, "string.replace"); + out_.reserve(size); + } else { + out_.reserve(s_.size()); + } + } + for (;;) { + if (sl.spent()) + return false; + if (in_to_) { + if (!append_part(out_, to_, off_, sl)) + return false; + in_to_ = false; + off_ = 0; + continue; + } + std::size_t at = npos; + if (scan_ < s_.size()) { + const std::size_t window = sl.bytes(s_.size() - scan_); + at = find_from(s_, from_, scan_, window); + if (at == npos) { + // No match starts before scan_ + window, so up to there `s` goes out as it is. + const std::size_t upto = std::min(s_.size(), scan_ + window); + out_.append(s_.substr(start_, upto - start_)); + sl.count_bytes(upto - scan_); + start_ = scan_ = upto; + continue; + } + } + if (at == npos) { + out_.append(s_.substr(start_)); + return true; + } + out_.append(s_.substr(start_, at - start_)); + sl.count_bytes(at + from_.size() - scan_); + start_ = scan_ = at + from_.size(); + in_to_ = true; + } + } + + value_t result() override { return value_t(std::move(out_)); } + +private: + std::string_view s_, from_, to_; + std::string s_store_, from_store_, to_store_; + bool sized_ = false; + match_counter matches_; + std::size_t start_ = 0, scan_ = 0; // what of `s` is out, and where the search goes on + bool in_to_ = false; // appending `to` for the match just found + std::size_t off_ = 0; + std::string out_; +}; + +// Copies a byte range (string.encode, bytes.slice) into a bytes value. +class copy_job final : public stdlib_job { +public: + copy_job(const char *who, std::string_view from) : stdlib_job(who), from_(from) { + out_.reserve(from.size()); + } + + void own_inputs() override { + own(from_, store_, off_); + off_ = 0; + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + return append_part(out_, from_, off_, sl); + } + + value_t result() override { return make_bytes(std::move(out_)); } + +private: + std::string_view from_; + std::string store_; + std::size_t off_ = 0; + std::string out_; +}; + +class concat_job final : public stdlib_job { +public: + concat_job(std::vector parts, std::size_t total) + : stdlib_job("bytes.concat"), parts_(std::move(parts)) { + out_.reserve(total); + } + + void own_inputs() override { + // What is still to be appended: the rest of the current part and the parts after it. + for (std::size_t i = next_; i < parts_.size(); ++i) + store_.emplace_back(parts_[i].substr(i == next_ ? std::min(off_, parts_[i].size()) : 0)); + parts_.assign(store_.begin(), store_.end()); + next_ = 0; + off_ = 0; + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; next_ < parts_.size(); ++next_, off_ = 0) + if (!append_part(out_, parts_[next_], off_, sl)) + return false; + return true; + } + + value_t result() override { return make_bytes(std::move(out_)); } + +private: + std::vector parts_; + std::vector store_; + std::size_t next_ = 0; + std::size_t off_ = 0; + std::string out_; +}; + +// bytes.decode: checks the bytes are well-formed UTF-8 as it copies them. +class decode_job final : public stdlib_job { +public: + explicit decode_job(std::string_view bytes) : stdlib_job("bytes.decode"), b_(bytes) { + out_.reserve(bytes.size()); + } + + void own_inputs() override { + own(b_, store_, pos_); + pos_ = 0; + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + while (pos_ < b_.size()) { + if (sl.spent()) + return false; + const std::size_t start = pos_, end = pos_ + sl.bytes(b_.size() - pos_); + while (pos_ < end) { + char32_t cp; + std::size_t len; + if (!utf8::decode(b_, pos_, cp, len)) + throw std::runtime_error("bytes.decode: invalid UTF-8"); + pos_ += len; + } + out_.append(b_.substr(start, pos_ - start)); + sl.count_bytes(pos_ - start); + } + return true; + } + + value_t result() override { return value_t(std::move(out_)); } + +private: + std::string_view b_; + std::string store_; + std::size_t pos_ = 0; + std::string out_; +}; + +// collections.map / filter: one call of `fn` per element. +class map_job final : public stdlib_job { +public: + map_job(const char *who, list_ptr list, value_t fn, bool filter) + : stdlib_job(who), list_(std::move(list)), fn_(std::move(fn)), filter_(filter) { + if (!filter_) + out_.reserve(list_->size()); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (i_ >= list_->size()) + return true; + if (sl.spent()) + return false; + arg_ = (*list_)[i_++]; + sl.count(1); + return call(fn_, {&arg_, 1}); + } + + void call_returned(value_t value) override { + if (!filter_) + out_.push_back(std::move(value)); + else if (value.is_truthy()) + out_.push_back(std::move(arg_)); + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + list_ptr list_; + value_t fn_; + bool filter_; + std::size_t i_ = 0; + value_t arg_; + std::vector out_; +}; + +class reduce_job final : public stdlib_job { +public: + reduce_job(list_ptr list, value_t fn, value_t init) + : stdlib_job("collections.reduce"), list_(std::move(list)), fn_(std::move(fn)), + acc_(std::move(init)) {} + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (i_ >= list_->size()) + return true; + if (sl.spent()) + return false; + args_[0] = std::move(acc_); + args_[1] = (*list_)[i_++]; + sl.count(1); + return call(fn_, {args_, 2}); + } + + void call_returned(value_t value) override { acc_ = std::move(value); } + + value_t result() override { return std::move(acc_); } + +private: + list_ptr list_; + value_t fn_; + value_t acc_; + std::size_t i_ = 0; + value_t args_[2]; +}; + +class zip_job final : public stdlib_job { +public: + zip_job(list_ptr a, list_ptr b, std::size_t n) + : stdlib_job("collections.zip"), a_(std::move(a)), b_(std::move(b)), n_(n) { + out_.reserve(n); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; i_ < n_ && i_ < a_->size() && i_ < b_->size(); ++i_) { + if (sl.spent()) + return false; + out_.push_back(make_list({(*a_)[i_], (*b_)[i_]})); + sl.count(1); + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + list_ptr a_, b_; + std::size_t n_; + std::size_t i_ = 0; + std::vector out_; +}; + +class flatten_job final : public stdlib_job { +public: + explicit flatten_job(list_ptr list) : stdlib_job("collections.flatten"), list_(std::move(list)) {} + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (!sized_) { + // The same sublist may appear many times, so the result can outgrow the argument: count + // the elements first. + for (; i_ < list_->size(); ++i_) { + if (sl.spent()) + return false; + const auto *sub = std::get_if(&(*list_)[i_].v); + total_ = sat_add(total_, sub ? (*sub ? (*sub)->size() : 0) : 1); + sl.count(1); + } + eval_alloc_check(list_bytes(total_), "collections.flatten"); + out_.reserve(total_); + sized_ = true; + i_ = 0; + } + for (; i_ < list_->size(); ++i_, j_ = 0) { + const value_t &item = (*list_)[i_]; + if (const auto *sub = std::get_if(&item.v)) { + const list_ptr &elems = *sub; + for (; elems && j_ < elems->size(); ++j_) { + if (sl.spent()) + return false; + add((*elems)[j_]); + sl.count(1); + } + } else { + if (sl.spent()) + return false; + add(item); + sl.count(1); + } + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + void add(const value_t &v) { + if (out_.size() >= total_) { // the list grew between slices + eval_alloc_check(list_bytes(out_.size() + 1), "collections.flatten"); + total_ = out_.size() + 1; + } + out_.push_back(v); + } + + list_ptr list_; + bool sized_ = false; + std::size_t total_ = 0; + std::size_t i_ = 0, j_ = 0; + std::vector out_; +}; + +// A bottom-up merge sort, one node per element moved: n log n nodes, and stable. +class sort_job final : public stdlib_job { +public: + explicit sort_job(list_ptr list) : stdlib_job("collections.sort"), list_(std::move(list)) {} + + bool step(std::size_t budget) override { + slice sl = begin(budget); + if (!copied_) { + for (; next_ < list_->size(); ++next_) { + if (sl.spent()) + return false; + a_.push_back((*list_)[next_]); + sl.count(1); + } + copied_ = true; + n_ = a_.size(); + if (n_ > 1) { + b_.resize(n_); + start_run(); + } + } + while (width_ < n_) { + if (k_ == hi_) { // this pair of runs is merged + lo_ = hi_; + if (lo_ >= n_) { // and so is this pass + std::swap(a_, b_); + width_ *= 2; + lo_ = 0; + if (width_ >= n_) + break; + } + start_run(); + continue; + } + if (sl.spent()) + return false; + if (i_ < mid_ && (j_ >= hi_ || !sort_less(a_[j_], a_[i_]))) + b_[k_++] = std::move(a_[i_++]); + else + b_[k_++] = std::move(a_[j_++]); + sl.count(1); + } + return true; + } + + value_t result() override { return make_list(std::move(a_)); } + +private: + void start_run() { + mid_ = std::min(lo_ + width_, n_); + hi_ = std::min(lo_ + 2 * width_, n_); + i_ = lo_; + j_ = mid_; + k_ = lo_; + } + + list_ptr list_; + bool copied_ = false; + std::size_t next_ = 0; + std::size_t n_ = 0; + std::vector a_, b_; // a_ is sorted in runs of width_; a pass merges them into b_ + std::size_t width_ = 1, lo_ = 0, mid_ = 0, hi_ = 0, i_ = 0, j_ = 0, k_ = 0; +}; + +class reverse_job final : public stdlib_job { +public: + explicit reverse_job(list_ptr list) + : stdlib_job("collections.reverse"), list_(std::move(list)), n_(list_->size()) { + out_.reserve(n_); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; k_ < n_; ++k_) { + if (sl.spent()) + return false; + if (const std::size_t i = n_ - 1 - k_; i < list_->size()) + out_.push_back((*list_)[i]); + sl.count(1); + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + list_ptr list_; + std::size_t n_; + std::size_t k_ = 0; + std::vector out_; +}; + +class range_job final : public stdlib_job { +public: + // Element k is start + k*step, in unsigned arithmetic (no signed overflow at the int64 ends). + range_job(uint64_t start, uint64_t step, std::size_t count) + : stdlib_job("collections.range"), start_(start), step_(step), count_(count) { + out_.reserve(count); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; k_ < count_; ++k_) { + if (sl.spent()) + return false; + out_.emplace_back(static_cast(start_ + k_ * step_)); + sl.count(1); + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + uint64_t start_, step_; + std::size_t count_; + std::size_t k_ = 0; + std::vector out_; +}; + +// Quadratic in the list length: one node per comparison, and a comparison of two lists or dicts +// runs as an equality walk sliced with the rest. +class unique_job final : public stdlib_job { +public: + explicit unique_job(list_ptr list) : stdlib_job("collections.unique"), list_(std::move(list)) {} + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; i_ < list_->size(); ++i_, j_ = 0) { + const value_t &item = (*list_)[i_]; + bool found = false; + while (j_ < out_.size()) { + if (sl.spent()) + return false; + bool equal; + if (cmp_) { + const std::size_t before = cmp_->nodes(); + const bool done = cmp_->step(sl.left()); + sl.count(std::max(1, cmp_->nodes() - before)); + if (!done) + return false; + equal = std::get(cmp_->result().v); + cmp_.reset(); + } else if (is_compound(out_[j_]) && is_compound(item)) { + cmp_ = make_equal_job(out_[j_], item); + continue; + } else { + equal = values_equal(out_[j_], item); + sl.count(1); + } + if (equal) { + found = true; + break; + } + ++j_; + } + if (!found) + out_.push_back(item); + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + list_ptr list_; + std::size_t i_ = 0, j_ = 0; + std::unique_ptr cmp_; // comparing out_[j_] with the element at i_ + std::vector out_; +}; + +// collections.dict-keys / dict-values. +class dict_items_job final : public stdlib_job { +public: + dict_items_job(const char *who, dict_ptr dict, bool keys) + : stdlib_job(who), dict_(std::move(dict)), keys_(keys) { + out_.reserve(dict_->size()); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; i_ < dict_->size(); ++i_) { + if (sl.spent()) + return false; + const auto &[k, v] = (*dict_)[i_]; + if (keys_) + out_.emplace_back(k); + else + out_.push_back(v); + sl.count(1); + } + return true; + } + + value_t result() override { return make_list(std::move(out_)); } + +private: + dict_ptr dict_; + bool keys_; + std::size_t i_ = 0; + std::vector out_; +}; + +// A copy of d1, then d2's entries: a key already there (its first entry) is updated in place, a +// new one appended. A key index makes it linear rather than quadratic. +class dict_merge_job final : public stdlib_job { +public: + dict_merge_job(dict_ptr d1, dict_ptr d2) + : stdlib_job("collections.dict-merge"), d1_(std::move(d1)), d2_(std::move(d2)), + out_(std::make_shared()) { + out_->reserve(d1_->size() + d2_->size()); + } + + bool step(std::size_t budget) override { + slice sl = begin(budget); + for (; i_ < d1_->size(); ++i_) { + if (sl.spent()) + return false; + index_.try_emplace((*d1_)[i_].first, out_->size()); + out_->push_back((*d1_)[i_]); + sl.count(1); + } + for (; j_ < d2_->size(); ++j_) { + if (sl.spent()) + return false; + const auto &[key, value] = (*d2_)[j_]; + if (auto [at, fresh] = index_.try_emplace(key, out_->size()); fresh) + out_->emplace_back(key, value); + else + (*out_)[at->second].second = value; + sl.count(1); + } + return true; + } + + value_t result() override { return value_t(std::move(out_)); } + +private: + dict_ptr d1_, d2_; + dict_ptr out_; + std::unordered_map index_; // key -> its entry in out_ + std::size_t i_ = 0, j_ = 0; +}; + } // anonymous namespace // --------------------------------------------------------------------------- @@ -86,62 +974,17 @@ void stdlib_registry::register_string_module() { register_function("string", "string.split", [](std::span args) -> value_t { expect_min(args, 1, "string.split"); auto &s = as_string(args[0], "string.split"); - std::string delim = " "; + std::string_view delim = " "; if (args.size() > 1) delim = as_string(args[1], "string.split"); - // Every part costs a list element, so many short parts outgrow the string: count first. - std::size_t count = 1; - if (delim.empty()) - count = utf8::count(s); - else - for (auto pos = s.find(delim); pos != std::string::npos; - pos = s.find(delim, pos + delim.size())) - ++count; - eval_alloc_check(sat_add(list_bytes(count), s.size()), "string.split"); - std::vector parts; - parts.reserve(count); - if (delim.empty()) { - // SE-1 (A3): split into CODEPOINTS, not bytes — each element is one whole UTF-8 codepoint, - // so a multibyte character is never shredded into invalid fragments. - std::size_t pos = 0; - while (pos < s.size()) { - char32_t cp; - std::size_t len; - utf8::decode(s, pos, cp, len); - parts.emplace_back(s.substr(pos, len)); - pos += len; - } - } else { - size_t start = 0; - while (start <= s.size()) { - auto pos = s.find(delim, start); - if (pos == std::string::npos) { - parts.emplace_back(s.substr(start)); - break; - } - parts.emplace_back(s.substr(start, pos - start)); - start = pos + delim.size(); - } - } - return make_list(std::move(parts)); + return run_data_job(std::make_unique(s, delim)); }); register_function("string", "string.join", [](std::span args) -> value_t { expect_exact(args, 2, "string.join"); - auto &lst = as_list(args[0], "string.join"); + list_ptr lst = as_list_ptr(args[0], "string.join"); auto &sep = as_string(args[1], "string.join"); - std::size_t total = lst.empty() ? 0 : alloc_bytes(lst.size() - 1, sep.size()); - for (const value_t &item : lst) - total = sat_add(total, as_string(item, "string.join").size()); - eval_alloc_check(total, "string.join"); - std::string result; - result.reserve(total); - for (size_t i = 0; i < lst.size(); ++i) { - if (i > 0) - result += sep; - result += as_string(lst[i], "string.join"); - } - return value_t(std::move(result)); + return run_data_job(std::make_unique(std::move(lst), sep)); }); register_function("string", "string.replace", [](std::span args) -> value_t { @@ -151,24 +994,7 @@ void stdlib_registry::register_string_module() { auto &to = as_string(args[2], "string.replace"); if (from.empty()) return value_t(s); - // Non-overlapping matches, left to right. A longer `to` multiplies the size: count first. - std::size_t count = 0; - for (auto pos = s.find(from); pos != std::string::npos; pos = s.find(from, pos + from.size())) - ++count; - if (count == 0) - return value_t(s); - const std::size_t size = sat_add(s.size() - count * from.size(), alloc_bytes(count, to.size())); - eval_alloc_check(size, "string.replace"); - std::string out; - out.reserve(size); - std::size_t start = 0; - for (auto pos = s.find(from); pos != std::string::npos; pos = s.find(from, start)) { - out.append(s, start, pos - start); - out += to; - start = pos + from.size(); - } - out.append(s, start); - return value_t(std::move(out)); + return run_data_job(std::make_unique(s, from, to)); }); register_function("string", "string.trim", [](std::span args) -> value_t { @@ -293,7 +1119,7 @@ void stdlib_registry::register_string_module() { if (enc != "utf-8" && enc != "utf8" && enc != "UTF-8") throw std::runtime_error("string.encode: unsupported encoding '" + enc + "' (only utf-8)"); } - return make_bytes(s); + return run_data_job(std::make_unique("string.encode", s)); }); } @@ -326,25 +1152,23 @@ void stdlib_registry::register_bytes_module() { pos = 0; if (static_cast(pos) >= b.size()) return make_bytes(std::string()); - if (args.size() > 2) { - int64_t len = as_int(args[2], "bytes.slice"); - if (len < 0) - len = 0; - return make_bytes(b.substr(static_cast(pos), static_cast(len))); - } - return make_bytes(b.substr(static_cast(pos))); + std::size_t len = npos; + if (args.size() > 2) + len = static_cast(std::max(0, as_int(args[2], "bytes.slice"))); + return run_data_job(std::make_unique( + "bytes.slice", std::string_view(b).substr(static_cast(pos), len))); }); register_function("bytes", "bytes.concat", [](std::span args) -> value_t { std::size_t total = 0; - for (const value_t &a : args) - total = sat_add(total, as_bytes(a, "bytes.concat").size()); + std::vector parts; + parts.reserve(args.size()); + for (const value_t &a : args) { + parts.emplace_back(as_bytes(a, "bytes.concat")); + total = sat_add(total, parts.back().size()); + } eval_alloc_check(total, "bytes.concat"); - std::string out; - out.reserve(total); - for (const value_t &a : args) - out += as_bytes(a, "bytes.concat"); - return make_bytes(std::move(out)); + return run_data_job(std::make_unique(std::move(parts), total)); }); // bytes.decode — binary->text. Only utf-8; the bytes MUST be valid UTF-8 (strict, like Python's @@ -357,9 +1181,8 @@ void stdlib_registry::register_bytes_module() { if (enc != "utf-8" && enc != "utf8" && enc != "UTF-8") throw std::runtime_error("bytes.decode: unsupported encoding '" + enc + "' (only utf-8)"); } - if (!utf8::is_valid(b)) - throw std::runtime_error("bytes.decode: invalid UTF-8"); - return value_t(b); // a string holding the validated UTF-8 bytes + // A string holding the validated UTF-8 bytes. + return run_data_job(std::make_unique(b)); }); } @@ -467,123 +1290,60 @@ void stdlib_registry::register_math_module() { // --------------------------------------------------------------------------- void stdlib_registry::register_collections_module() { + // map, filter and reduce take a native function or a DSL function (a lambda, a defun): the + // evaluator running the program makes each call, a DSL one as its own frames. register_function("collections", "collections.map", [](std::span args) -> value_t { expect_exact(args, 2, "collections.map"); - auto &lst = as_list(args[0], "collections.map"); - auto *fn = std::get_if(&args[1].v); - if (!fn) - throw std::runtime_error("collections.map: second argument must be a function"); - std::vector result; - result.reserve(lst.size()); - for (auto &item : lst) { - result.push_back((*fn)(std::span(&item, 1))); - } - return make_list(std::move(result)); + list_ptr lst = as_list_ptr(args[0], "collections.map"); + const value_t &fn = as_function(args[1], "collections.map", "second"); + return run_data_job(std::make_unique("collections.map", std::move(lst), fn, false)); }); register_function( "collections", "collections.filter", [](std::span args) -> value_t { expect_exact(args, 2, "collections.filter"); - auto &lst = as_list(args[0], "collections.filter"); - auto *fn = std::get_if(&args[1].v); - if (!fn) - throw std::runtime_error("collections.filter: second argument must be a function"); - std::vector result; - for (auto &item : lst) { - auto v = (*fn)(std::span(&item, 1)); - if (v.is_truthy()) - result.push_back(item); - } - return make_list(std::move(result)); + list_ptr lst = as_list_ptr(args[0], "collections.filter"); + const value_t &fn = as_function(args[1], "collections.filter", "second"); + return run_data_job( + std::make_unique("collections.filter", std::move(lst), fn, true)); }); register_function( "collections", "collections.reduce", [](std::span args) -> value_t { expect_exact(args, 3, "collections.reduce"); - auto &lst = as_list(args[0], "collections.reduce"); - auto *fn = std::get_if(&args[1].v); - if (!fn) - throw std::runtime_error("collections.reduce: second argument must be a function"); - value_t acc = args[2]; - for (auto &item : lst) { - value_t pair[2] = {acc, item}; - acc = (*fn)(std::span(pair, 2)); - } - return acc; + list_ptr lst = as_list_ptr(args[0], "collections.reduce"); + const value_t &fn = as_function(args[1], "collections.reduce", "second"); + return run_data_job(std::make_unique(std::move(lst), fn, args[2])); }); register_function("collections", "collections.zip", [](std::span args) -> value_t { expect_exact(args, 2, "collections.zip"); - auto &a = as_list(args[0], "collections.zip"); - auto &b = as_list(args[1], "collections.zip"); - size_t n = std::min(a.size(), b.size()); + list_ptr a = as_list_ptr(args[0], "collections.zip"); + list_ptr b = as_list_ptr(args[1], "collections.zip"); + const std::size_t n = std::min(a->size(), b->size()); eval_alloc_check(sat_add(list_bytes(n), alloc_bytes(n, list_bytes(2))), "collections.zip"); - std::vector result; - result.reserve(n); - for (size_t i = 0; i < n; ++i) { - result.push_back(make_list({a[i], b[i]})); - } - return make_list(std::move(result)); + return run_data_job(std::make_unique(std::move(a), std::move(b), n)); }); register_function("collections", "collections.flatten", [](std::span args) -> value_t { expect_exact(args, 1, "collections.flatten"); - auto &lst = as_list(args[0], "collections.flatten"); - // The same sublist may appear many times, so the result can outgrow the - // argument: count the elements first. - std::size_t total = 0; - for (auto &item : lst) { - auto *sub = std::get_if(&item.v); - total = sat_add(total, sub ? (**sub).size() : 1); - } - eval_alloc_check(list_bytes(total), "collections.flatten"); - std::vector result; - result.reserve(total); - for (auto &item : lst) { - if (auto *sub = std::get_if(&item.v)) { - for (auto &x : **sub) - result.push_back(x); - } else { - result.push_back(item); - } - } - return make_list(std::move(result)); + return run_data_job(std::make_unique( + as_list_ptr(args[0], "collections.flatten"))); }); register_function("collections", "collections.sort", [](std::span args) -> value_t { expect_exact(args, 1, "collections.sort"); - auto elems = as_list(args[0], "collections.sort"); // copy - std::sort(elems.begin(), elems.end(), [](const value_t &a, const value_t &b) { - // Sort by numeric value or string comparison - if (auto *ai = std::get_if(&a.v)) { - if (auto *bi = std::get_if(&b.v)) - return *ai < *bi; - if (auto *bd = std::get_if(&b.v)) - return static_cast(*ai) < *bd; - } - if (auto *ad = std::get_if(&a.v)) { - if (auto *bi = std::get_if(&b.v)) - return *ad < static_cast(*bi); - if (auto *bd = std::get_if(&b.v)) - return *ad < *bd; - } - if (auto *as = std::get_if(&a.v)) { - if (auto *bs = std::get_if(&b.v)) - return *as < *bs; - } - return false; - }); - return make_list(std::move(elems)); + return run_data_job( + std::make_unique(as_list_ptr(args[0], "collections.sort"))); }); register_function("collections", "collections.reverse", [](std::span args) -> value_t { expect_exact(args, 1, "collections.reverse"); - auto elems = as_list(args[0], "collections.reverse"); // copy - std::reverse(elems.begin(), elems.end()); - return make_list(std::move(elems)); + return run_data_job(std::make_unique( + as_list_ptr(args[0], "collections.reverse"))); }); register_function( @@ -612,86 +1372,42 @@ void stdlib_registry::register_collections_module() { count = (ustart - ustop - 1) / (0 - ustep) + 1; eval_alloc_check(count > SIZE_MAX ? SIZE_MAX : list_bytes(static_cast(count)), "collections.range"); - std::vector result; - result.reserve(static_cast(count)); - for (uint64_t k = 0; k < count; ++k) - result.emplace_back(static_cast(ustart + k * ustep)); - return make_list(std::move(result)); + return run_data_job( + std::make_unique(ustart, ustep, static_cast(count))); }); register_function("collections", "collections.unique", [](std::span args) -> value_t { expect_exact(args, 1, "collections.unique"); - auto &lst = as_list(args[0], "collections.unique"); - std::vector result; - for (auto &item : lst) { - poll_deadline("collections.unique"); // quadratic in the list length - bool found = false; - for (auto &r : result) { - if (values_equal(r, item)) { - found = true; - break; - } - } - if (!found) - result.push_back(item); - } - return make_list(std::move(result)); + return run_data_job( + std::make_unique(as_list_ptr(args[0], "collections.unique"))); }); register_function("collections", "collections.dict-keys", [](std::span args) -> value_t { expect_exact(args, 1, "collections.dict-keys"); - auto *d = std::get_if(&args[0].v); - if (!d) - throw std::runtime_error("collections.dict-keys: expected dict"); - std::vector keys; - keys.reserve((**d).size()); - for (auto &[k, _] : **d) - keys.emplace_back(k); - return make_list(std::move(keys)); + dict_ptr d = as_dict_ptr(args[0], "collections.dict-keys", "expected dict"); + return run_data_job(std::make_unique("collections.dict-keys", + std::move(d), true)); }); register_function("collections", "collections.dict-values", [](std::span args) -> value_t { expect_exact(args, 1, "collections.dict-values"); - auto *d = std::get_if(&args[0].v); - if (!d) - throw std::runtime_error("collections.dict-values: expected dict"); - std::vector vals; - vals.reserve((**d).size()); - for (auto &[_, v] : **d) - vals.push_back(v); - return make_list(std::move(vals)); + dict_ptr d = as_dict_ptr(args[0], "collections.dict-values", "expected dict"); + return run_data_job(std::make_unique( + "collections.dict-values", std::move(d), false)); }); register_function( "collections", "collections.dict-merge", [](std::span args) -> value_t { expect_exact(args, 2, "collections.dict-merge"); - auto *d1 = std::get_if(&args[0].v); - auto *d2 = std::get_if(&args[1].v); - if (!d1 || !d2) - throw std::runtime_error("collections.dict-merge: expected two dicts"); + dict_ptr d1 = as_dict_ptr(args[0], "collections.dict-merge", "expected two dicts"); + dict_ptr d2 = as_dict_ptr(args[1], "collections.dict-merge", "expected two dicts"); eval_alloc_check( - alloc_bytes((**d1).size() + (**d2).size(), sizeof(std::pair)), + alloc_bytes(d1->size() + d2->size(), sizeof(std::pair)), "collections.dict-merge"); - // Start with copy of d1 - auto result = std::make_shared>>(**d1); - // Merge d2: update existing keys, append new ones - for (auto &[k, v] : **d2) { - poll_deadline("collections.dict-merge"); // quadratic in the dict sizes - bool found = false; - for (auto &[rk, rv] : *result) { - if (rk == k) { - rv = v; - found = true; - break; - } - } - if (!found) - result->emplace_back(k, v); - } - return value_t(std::move(result)); + return run_data_job(std::make_unique(std::move(d1), std::move(d2))); }); } diff --git a/src/cvc/tests/ariadne_runtime_test.cpp b/src/cvc/tests/ariadne_runtime_test.cpp index 67e236eea..ea7a03dc5 100644 --- a/src/cvc/tests/ariadne_runtime_test.cpp +++ b/src/cvc/tests/ariadne_runtime_test.cpp @@ -1622,28 +1622,127 @@ TEST(AriadneStdlib, PredicateGetsThePureBoundedSubset) { EXPECT_TRUE(rt.take_reactive_warnings().empty()); } -TEST(AriadneStdlib, PredicateExcludesStdlibMaterializersAndRandom) { +// The materializers run in a predicate as sliced data steps, and collections.map/filter/reduce +// take a lambda there; math.random stays out (a predicate must be deterministic). +TEST(AriadneStdlib, PredicateRunsTheMaterializersButNotRandom) { + cvc::app app; + Runtime rt(app, ""); + MockBackend mb; + rt.set_backend(&mb); + ASSERT_TRUE(run_init(app, "", "(state-set \"csv\" \"b,a,c,a\")", nullptr)); + const std::vector> shown = { + {"split", "(= (length (string.split (state-get \"csv\") \",\")) 4)"}, + {"join", "(= (string.join (collections.sort (string.split (state-get \"csv\") \",\")) \"\") " + "\"aabc\")"}, + {"replace", "(= (string.replace (state-get \"csv\") \",\" \"\") \"baca\")"}, + {"bytes", "(= (bytes.decode (bytes.concat (bytes.slice (string.encode \"xab\") 1) " + "(string.encode \"!\"))) \"ab!\")"}, + {"map", "(= (collections.map (list 1 2 3) (lambda (x) (* x x))) (list 1 4 9))"}, + {"filter", "(= (length (collections.filter (collections.range 10) (lambda (x) (> x 6)))) 3)"}, + {"reduce", "(= (collections.reduce (collections.range 5) (lambda (a x) (+ a x)) 0) 10)"}, + {"unique", "(= (collections.unique (string.split (state-get \"csv\") \",\")) " + "(list \"b\" \"a\" \"c\"))"}, + {"dicts", "(= (collections.dict-keys (collections.dict-merge (dict \"a\" 1) (dict \"b\" 2))) " + "(list \"a\" \"b\"))"}, + {"shape", "(= (collections.flatten (collections.zip (list 1 2) (collections.reverse (list 3 " + "4)))) (list 1 4 2 3))"}}; + std::vector widgets; + for (const auto &[name, pred] : shown) { + widgets.push_back(text(name)); + widgets.back().visible_when = pred; + } + Widget rnd = text("rnd"); + rnd.visible_when = "(< (math.random) 2)"; // unbound -> fail-safe hidden + widgets.push_back(rnd); + rt.set_root(group(widgets)); + rt.render(); + for (const auto &[name, pred] : shown) + EXPECT_TRUE(mb.saw(std::string("text_line:") + name)) << pred; + EXPECT_FALSE(mb.saw("text_line:rnd")); + const auto warns = rt.take_reactive_warnings(); + ASSERT_EQ(warns.size(), 1u) << (warns.empty() ? "" : warns[0]); + EXPECT_NE(warns[0].find("math.random"), std::string::npos) << warns[0]; +} + +// A predicate's deep-data work runs a slice per step, so its step and time caps cut a huge +// materializer instead of the frame waiting for it. data.step_nodes comes from the app tree. +TEST(AriadneStdlib, PredicateMaterializerIsCutByTheEvalBudget) { cvc::app app; Runtime rt(app, ""); MockBackend mb; rt.set_backend(&mb); - // Results that can outgrow their arguments wait for scheduled, sliced versions; math.random - // would make a predicate nondeterministic. Each is an unbound symbol -> fail-safe hidden. - Widget join = text("join"); - join.visible_when = "(is-string (string.join (list \"a\") \",\"))"; - Widget rep = text("rep"); - rep.visible_when = "(is-string (string.replace \"a\" \"a\" \"aa\"))"; - Widget range = text("range"); - range.visible_when = "(is-list (collections.range 3))"; - Widget rnd = text("rnd"); - rnd.visible_when = "(< (math.random) 2)"; - rt.set_root(group({join, rep, range, rnd})); + cvc::state::instance(app)("state_exec.defaults.data.step_nodes").value("1"); + Widget big = text("big"); + big.visible_when = "(> (length (collections.range 120000)) 0)"; // > 100000 one-node steps + rt.set_root(group({big})); rt.render(); - EXPECT_FALSE(mb.saw("text_line:join")); - EXPECT_FALSE(mb.saw("text_line:rep")); - EXPECT_FALSE(mb.saw("text_line:range")); - EXPECT_FALSE(mb.saw("text_line:rnd")); + EXPECT_FALSE(mb.saw("text_line:big")); + auto warns = rt.take_reactive_warnings(); + ASSERT_FALSE(warns.empty()); + EXPECT_NE(warns[0].find("exceeded the eval budget"), std::string::npos) << warns[0]; + + // A 0 here still slices, at the host's 1024 nodes a step, and an ordinary call finishes. It is + // kept small so it fits the 5 ms time cap in a Debug build on a loaded CI runner. + cvc::state::instance(app)("state_exec.defaults.data.step_nodes").value("0"); + Widget fits = text("fits"); + fits.visible_when = "(= (length (collections.range 1000)) 1000)"; + rt.set_root(group({fits})); + MockBackend mb2; + rt.set_backend(&mb2); + rt.render(); + EXPECT_TRUE(mb2.saw("text_line:fits")); + EXPECT_TRUE(rt.take_reactive_warnings().empty()); +} + +// A lambda a predicate passes to collections.map runs on the predicate's own evaluator: the +// special-form gate, the reduced environment and the caps all hold inside it. +TEST(AriadneStdlib, PredicateLambdasStayInsideThePredicateRules) { + cvc::app app; + Runtime rt(app, ""); + MockBackend mb; + rt.set_backend(&mb); + Widget macro = text("macro"); + macro.visible_when = "(collections.map (list 1) (lambda (x) (defmacro m () 1)))"; + rt.set_root(group({macro})); + rt.render(); + EXPECT_FALSE(mb.saw("text_line:macro")); + auto warns = rt.take_reactive_warnings(); + ASSERT_FALSE(warns.empty()); + EXPECT_NE(warns[0].find("not permitted"), std::string::npos) << warns[0]; + + Widget write = text("write"); + write.visible_when = "(collections.map (list 1) (lambda (x) (state-set \"leak\" \"1\")))"; + rt.set_root(group({write})); + rt.render(); + EXPECT_FALSE(mb.saw("text_line:write")); EXPECT_FALSE(rt.take_reactive_warnings().empty()); + EXPECT_NE(cvc::state::instance(app)("leak").value(), "1"); + + Widget spin = text("spin"); + spin.visible_when = "(collections.map (list 1) (lambda (x) (while t 1)))"; + rt.set_root(group({spin})); + rt.render(); + EXPECT_FALSE(mb.saw("text_line:spin")); + warns = rt.take_reactive_warnings(); + ASSERT_FALSE(warns.empty()); + EXPECT_NE(warns[0].find("exceeded the eval budget"), std::string::npos) << warns[0]; +} + +TEST(AriadneStdlib, ActionLaneCallsLambdasThroughMapAndApply) { + cvc::app app; + Runtime rt(app, "ui"); + MockBackend mb; + rt.set_backend(&mb); + rt.set_root(group({button("Go", "(begin (state-set \"out\" (string.join (collections.map " + "(string.split \"a b c\" \" \") (lambda (w) (string.upper w))) " + "\"-\")) (state-set \"sum\" (apply (lambda (a b) (+ a b)) " + "(list 2 3))))")})); + mb.button_click = true; + rt.render(); + rt.drain(); + EXPECT_EQ(cvc::state::instance(app)("ui.out").value(), "A-B-C"); + EXPECT_EQ(cvc::state::instance(app)("ui.sum").value(), "5"); + EXPECT_TRUE(rt.take_reactive_warnings().empty()); } TEST(AriadneStdlib, InitAndActionLanesReachTheWholeStdlib) { diff --git a/src/cvc/tests/state_exec_bench_test.cpp b/src/cvc/tests/state_exec_bench_test.cpp index f089768cd..f973adeb3 100644 --- a/src/cvc/tests/state_exec_bench_test.cpp +++ b/src/cvc/tests/state_exec_bench_test.cpp @@ -31,6 +31,7 @@ #include #include #include +#include namespace { @@ -230,6 +231,65 @@ TEST_F(ExecBenchTest, DataWalkThroughput) { std::chrono::duration_cast(t1 - t0)); } +TEST_F(ExecBenchTest, StdlibJobThroughput) { + if (!bench_enabled()) { + SUCCEED(); + return; + } + + // The stdlib materializers' work per node (one per element, one per kDataNodeBytes bytes + // scanned or copied), run in 1024-node data steps on a stackless evaluator: a node's cost + // includes the stepping, and freeing the result. + constexpr int64_t kN = 200000; + auto env = builtins::make_default_environment(); + auto run_env = environment::extend(env); + std::string csv; + std::vector nums, parts, rev; + for (int64_t i = 0; i < kN; ++i) { + csv += (i ? "," : "") + std::to_string(i); + nums.emplace_back(i); + parts.emplace_back(std::to_string(i)); + rev.emplace_back(kN - i); + } + run_env->set("csv", value_t{csv}); + run_env->set("nums", make_list(std::move(nums))); + run_env->set("parts", make_list(std::move(parts))); + run_env->set("rev", make_list(std::move(rev))); + const auto per_node = [&](const char *name, const char *script) { + stackless_evaluator ev(env); + evaluator_state st = ev.create_state(parse(script), run_env); + st.data_step_nodes = 1024; + auto t0 = hrclock::now(); + while (!ev.step(st)) { + } + auto t1 = hrclock::now(); + report(name, static_cast(st.stats.data_nodes()), + std::chrono::duration_cast(t1 - t0)); + }; + per_node("stdlib_split_per_node", "(string.split csv \",\")"); + per_node("stdlib_join_per_node", "(string.join parts \",\")"); + per_node("stdlib_replace_per_node", "(string.replace csv \",\" \";\")"); + per_node("stdlib_encode_per_node", "(string.encode csv)"); + per_node("stdlib_sort_per_node", "(collections.sort rev)"); + per_node("stdlib_map_native_per_node", "(collections.map nums math.abs)"); + per_node("stdlib_map_lambda_per_node", "(collections.map nums (lambda (x) x))"); + per_node("stdlib_dict_merge_per_node", + "(collections.dict-merge (dict \"a\" 1 \"b\" 2) (dict \"b\" 3 \"c\" 4))"); + + // Small calls, where a job's setup is most of the cost. + const std::string loop = "(begin (set i 0) (while (< i 10000) (begin" + " (string.split \"a,b,c\" \",\") (collections.reverse (list 1 2 3))" + " (set i (+ i 1)))) i)"; + constexpr int M = 20; + auto t0 = hrclock::now(); + for (int i = 0; i < M; ++i) { + stackless_evaluator ev(env); + ev.evaluate_script(loop); + } + auto t1 = hrclock::now(); + report("stdlib_small_calls_10000", M, std::chrono::duration_cast(t1 - t0)); +} + TEST_F(ExecBenchTest, SchedulerThroughput) { if (!bench_enabled()) { SUCCEED(); diff --git a/src/cvc/tests/state_exec_data_steps_test.cpp b/src/cvc/tests/state_exec_data_steps_test.cpp index 883dbc2b6..db5dfc6d2 100644 --- a/src/cvc/tests/state_exec_data_steps_test.cpp +++ b/src/cvc/tests/state_exec_data_steps_test.cpp @@ -15,8 +15,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -304,6 +306,119 @@ TEST_F(DataStepsEvaluator, SnapshotMidWorkRunsTheCallAgain) { EXPECT_GT(back.stats.steps(step_kind::data), 0u); } +// ---- jobs that call functions ----------------------------------------------------- + +// A job asks for a call and the evaluator makes it: a native function in the data step, a DSL +// function as frames above the call waiting on the job, so its body runs as the program's own +// evaluation steps. +TEST_F(DataStepsEvaluator, ALambdaAJobCallsRunsAsFramesAboveTheCall) { + evaluator_state st = ev.create_state( + parse("(+ 1 (length (collections.map (collections.range 40) (lambda (x) (* x 2)))))")); + st.data_step_nodes = 4; + st.stats.start(); + bool saw_call = false; + while (!ev.step(st)) { + // While the lambda runs its frames are on top; once it returns (stack back at base) the + // next step hands its value to the job. + if (st.data_work && st.data_work->calling && st.stack.size() > st.data_work->base) { + saw_call = true; + EXPECT_EQ(st.stack[st.data_work->base - 1].phase, eval_phase::apply); // the waiting call + } + } + EXPECT_TRUE(saw_call); + EXPECT_EQ(std::get(st.result.v), 41); + EXPECT_GE(st.stats.steps(step_kind::data), 40u); +} + +TEST_F(DataStepsEvaluator, SnapshotWhileAJobsLambdaRunsRunsTheCallAgain) { + cvc::app ctx; + evaluator_state st = ev.create_state( + parse("(+ 1 (length (collections.map (collections.range 40) (lambda (x) (* x 2)))))")); + st.data_step_nodes = 4; + st.stats.start(); + while (!(st.data_work && st.data_work->calling && st.stack.size() > st.data_work->base && + st.stats.steps(step_kind::data) >= 10)) + ASSERT_FALSE(ev.step(st)); + const std::size_t base = st.data_work->base; + const std::size_t args = st.data_work->frame_args; + ASSERT_GT(st.stack.size(), base); + + auto &node = cvc::state::instance(ctx)("snapshot"); + encode_evaluator_state(node, st); + evaluator_state back = decode_evaluator_state(node); + EXPECT_EQ(back.data_work, nullptr); + EXPECT_EQ(back.stack.size(), base); // the lambda's frames are left out... + EXPECT_EQ(back.stack.back().results.size(), args); // ...and the waiting call keeps its args + // Native functions are not serialized: give each decoded scope chain the builtins and the + // stdlib again. + const auto patch = [&](environment_ptr e) { + for (; e && e->outer && !e->outer->frozen; e = e->outer) { + } + if (!e || e->outer) + return; // no scope, or patched already + for (auto &[name, val] : env->bindings) + e->set(name, val); + e->outer = stdlib_registry::shared_scope(); + }; + patch(back.global_env); + for (auto &f : back.stack) { + patch(f.env); + for (auto &v : f.results) + if (auto *c = std::get_if(&v.v)) + patch((*c)->env_snapshot); + } + ev.run(back); // the map runs again from its first element + ASSERT_TRUE(back.done); + EXPECT_EQ(std::get(back.result.v), 41); + ev.run(st); + EXPECT_EQ(std::get(st.result.v), 41); +} + +// A native function a job calls may leave deep-data work of its own (str of a value bigger than +// the slice): its result is that call's value, not the whole job's. Before jobs made their calls +// through the evaluator, (collections.map (list a b) str) came back as one string. +TEST_F(DataStepsEvaluator, ANativeFunctionsOwnWorkIsItsCallsValue) { + const std::string expected = + "(\"" + to_string(nested_list(1000)) + "\" \"" + to_string(nested_list(1000)) + "\")"; + for (const char *script : + {"(collections.map (list a b) str)", "(collections.filter (list a b) (lambda (x) (= x b)))", + "(collections.reduce (list a b) (lambda (acc x) (cons (str x) acc)) " + "(list))"}) { + SCOPED_TRACE(script); + evaluator_state whole = state(script, 0); + step_to_done(ev, whole); + evaluator_state sliced = state(script, 16); + step_to_done(ev, sliced); + EXPECT_EQ(to_string(sliced.result), to_string(whole.result)); + EXPECT_GT(sliced.stats.steps(step_kind::data), 10u); + } + evaluator_state mapped = state("(collections.map (list a b) str)", 16); + step_to_done(ev, mapped); + ASSERT_TRUE(std::holds_alternative(mapped.result.v)); + EXPECT_EQ(std::get(mapped.result.v)->size(), 2u); + EXPECT_EQ(to_string(mapped.result), expected); +} + +TEST_F(DataStepsEvaluator, ALambdaAJobCallsIsGatedLikeTheProgram) { + // The special-form gate of a restricted evaluator (Ariadne's predicates) covers the body of a + // function a job calls, since that body runs on the same evaluator. + stackless_evaluator gated(env); + gated.restrict_special_forms(std::make_shared>( + std::set{"lambda", "if", "quote"})); + for (std::size_t slice : {0u, 16u}) { + evaluator_state st = + gated.create_state(parse("(collections.map (list 1) (lambda (x) (defmacro m () 1)))")); + st.data_step_nodes = slice; + std::string err; + try { + gated.run(st); + } catch (const std::runtime_error &e) { + err = e.what(); + } + EXPECT_NE(err.find("special form 'defmacro' is not permitted"), std::string::npos) << err; + } +} + // ---- the schedulers ------------------------------------------------------------- class DataStepsScheduler : public ::testing::Test { @@ -387,6 +502,19 @@ TEST_F(DataStepsScheduler, OtherProcessesRunBetweenSlices) { EXPECT_TRUE(std::get(sched.get_result(slow)->v)); } +TEST_F(DataStepsScheduler, NoForkWhileAJobsFunctionRuns) { + // The job is not copied, so a child could not finish the call: fork refuses until it is done. + scheduler sched; + const int pid = + sched.execute(std::string("(collections.map (collections.range 50) (lambda (x) (* x 2)))")); + while (sched.get_process_info(pid)->data_steps < 5) + sched.step(); + EXPECT_EQ(sched.fork(pid), -1); + sched.run(); + EXPECT_EQ(sched.get_process_info(pid)->status, process_status::terminated); + EXPECT_EQ(sched.list_processes().size(), 1u); +} + TEST_F(DataStepsScheduler, SignalsWaitForDeepWork) { root("state_exec.defaults.data.step_nodes").value("4"); scheduler sched; diff --git a/src/cvc/tests/state_exec_heap_test.cpp b/src/cvc/tests/state_exec_heap_test.cpp index 865306573..fff4613fc 100644 --- a/src/cvc/tests/state_exec_heap_test.cpp +++ b/src/cvc/tests/state_exec_heap_test.cpp @@ -8,8 +8,11 @@ #include #include #include +#include #include #include +#include +#include #include #include #include @@ -363,3 +366,91 @@ TEST(HeapBudgetTest, AsyncSchedulerCountsTheHeapToo) { error.find("allocation limit") != std::string::npos) << error; } + +// --- a measurement next to the evaluator's other deep-data work ------------------------------ + +// A measurement starts between two steps and waits on no frame: it records the whole stack as +// its base, so a snapshot taken while it runs keeps every frame, and evaluation resumes where it +// was. (The other deep-data work, a native call's job, waits on its call's frame instead.) +TEST(HeapBudgetTest, AMeasurementWaitsOnNoFrame) { + cvc::app app; + auto env = builtins::make_default_environment(); + env->set("keep", value_t{int_list(2000)}); + stackless_evaluator ev(env); + process p; + p.max_memory = std::size_t{64} << 20; + p.state = ev.create_state(parse("(+ 1 (length keep) (length keep))")); + p.state.global_env = env; + p.state.data_step_nodes = 8; + p.state.stats.start(); + for (int i = 0; i < 4; ++i) + ASSERT_FALSE(ev.step(p.state)); + const std::size_t frames = p.state.stack.size(); + ASSERT_GT(frames, 1u); + for (int i = 0; i < 64 && !p.state.data_work; ++i) + heap_after_step(p, std::size_t{1} << 20, 0); // allocations call for a measurement + ASSERT_TRUE(p.state.data_work); + ASSERT_EQ(p.state.data_work->what, pending_data_work::kind::heap_measure); + EXPECT_EQ(p.state.data_work->base, frames); + EXPECT_EQ(p.state.data_work->frame_args, p.state.stack.back().results.size()); + + auto &node = cvc::state::instance(app)("snapshot"); + encode_evaluator_state(node, p.state); + EXPECT_EQ(decode_evaluator_state(node).stack.size(), frames); + // Even a measurement made without a base (0) never shortens a snapshot's stack. + p.state.data_work->base = 0; + auto &again = cvc::state::instance(app)("snapshot2"); + encode_evaluator_state(again, p.state); + EXPECT_EQ(decode_evaluator_state(again).stack.size(), frames); + p.state.data_work->base = frames; + + const int steps = [&] { + int n = 0; + while (!ev.step(p.state)) + ++n; + return n; + }(); + EXPECT_GT(steps, 0); + ASSERT_TRUE(p.state.heap_measured); + EXPECT_EQ(std::get(p.state.result.v), 4001); +} + +// fork refuses while deep-data work a call waits on is in flight, but a measurement holds no +// frame, so a process can fork in the middle of one. +TEST(HeapBudgetTest, AProcessForksDuringAMeasurement) { + cvc::app app; + auto &root = cvc::state::instance(app); + // Small slices, so a measurement spans many steps; a measurement every 100 steps. + root("state_exec.defaults.data.step_nodes").value(std::string("4")); + root("state_exec.defaults.limits.heap_check_steps").value(std::string("100")); + scheduler sched; + sched.set_watch_root(&root); + sched.load_settings(); + execute_options opts; + opts.max_memory = std::size_t{64} << 20; + const int pid = + sched.execute(std::string("(begin (set keep (list)) (set i 0) (while (< i 400) " + "(begin (append keep (list i i)) (set i (+ i 1)))) (length keep))"), + opts); + // The program has no other deep-data work, so a data step that does not finish a measurement + // is one in progress (or, rarely, freeing). Fork there, three times. + process_info before = *sched.get_process_info(pid); + int forks = 0; + while (forks < 3 && before.status != process_status::terminated) { + sched.step(); + const process_info now = *sched.get_process_info(pid); + if (now.data_steps > before.data_steps && now.heap_measurements == before.heap_measurements && + now.status != process_status::terminated) { + const int child = sched.fork(pid); + EXPECT_GE(child, 0); + ++forks; + if (child >= 0) + sched.kill(child); + } + before = now; + } + EXPECT_EQ(forks, 3); + sched.run(); + EXPECT_GT(sched.get_process_info(pid)->heap_measurements, 0u); + EXPECT_EQ(std::get(sched.get_result(pid)->v), 400); +} diff --git a/src/cvc/tests/state_exec_stdlib_test.cpp b/src/cvc/tests/state_exec_stdlib_test.cpp index 6e93cd8a9..150170b99 100644 --- a/src/cvc/tests/state_exec_stdlib_test.cpp +++ b/src/cvc/tests/state_exec_stdlib_test.cpp @@ -1,16 +1,21 @@ /// @file state_exec_stdlib_test.cpp /// @brief The stdlib modules (string, bytes, math, collections) reached from DSL programs, the -/// shared frozen scope they live in, and the per-call allocation limit. +/// shared frozen scope they live in, the per-call allocation limit, and the materializers' work +/// as sliced data steps. #include #include #include #include +#include #include #include +#include #include +#include #include #include +#include #include using namespace cvc::state_exec; @@ -359,3 +364,269 @@ TEST(StdlibBehaviourTest, FlattenAndZip) { EXPECT_EQ(run_sync("(collections.flatten (list (list 1 2) 3 (list 4)))"), "(1 2 3 4)"); EXPECT_EQ(run_sync("(collections.zip (list 1 2 3) (list \"a\" \"b\"))"), "((1 \"a\") (2 \"b\"))"); } + +// --- sliced work on the executor ------------------------------------------------------------- +// The materializers (string.split/join/replace/encode, bytes.slice/concat/decode, collections.*) +// do their work as data jobs: a slice per scheduler step (data.step_nodes), counted against +// max_steps/max_time as data steps, and the functions collections.map/filter/reduce call run as +// the process's own frames. + +namespace { + +// A sync scheduler slicing deep-data work `slice` nodes a step (0 = not at all). +struct sliced_scheduler { + cvc::app app; + cvc::state &root = cvc::state::instance(app); + scheduler sched; + + explicit sliced_scheduler(std::size_t slice) { + root("state_exec.defaults.data.step_nodes").value(std::to_string(slice)); + sched.set_watch_root(&root); + sched.load_settings(); + } + + // Run `script` to completion: its result shown, and its process's info. + std::pair run(const std::string &script, execute_options opts = {}) { + const int pid = sched.execute(script, opts); + sched.run(); + return {show(sched.get_result(pid)), *sched.get_process_info(pid)}; + } +}; + +// Each sliced function, and the ways a job calls a function, on inputs that need many slices +// of a few nodes. +const char *const kSlicedScripts[] = { + "(string.split \"a,b,,c,\" \",\")", + "(string.split \"one two three\")", + "(string.split \"\" \",\")", + "(string.split \"caf\xC3\xA9 \xE2\x82\xAC\" \"\")", + "(string.split (string.join (collections.map (collections.range 500) str) \"::\") \"::\")", + "(string.join (collections.map (collections.range 400) str) \", \")", + "(string.replace (string.join (collections.map (collections.range 300) str) \"-\") \"1\" " + "\"\")", + "(string.replace \"aaaa\" \"aa\" \"b\")", + "(string.replace \"abc\" \"x\" \"y\")", + "(bytes.length (string.encode (string.join (collections.map (collections.range 3000) str) " + "\"\")))", + "(bytes.decode (bytes.slice (string.encode \"hello, w\xC3\xB6rld\") 7))", + "(bytes.decode (bytes.concat (string.encode \"ab\") (string.encode \"\") (string.encode " + "\"cd\")))", + "(collections.map (collections.range 60) (lambda (x) (* x x)))", + "(collections.map (list (list 1 2) (list 3)) (lambda (xs) (collections.map xs (lambda (x) (* " + "10 x)))))", + "(collections.map (list (collections.range 300) (list 1)) str)", + "(collections.map (list 1 -2 3) (lambda (x) (if (< x 0) (return 0) x)))", + "(collections.map (list 2 3) (lambda (n) (begin (set i 0) (while t (begin (set i (+ i 1)) " + "(if (= i n) (break (* i 10)) nil))))))", + "(collect (generator (lambda () (yield (collections.map (list 1 2) (lambda (x) (+ x 1)))))))", + "(collections.filter (collections.range 200) (lambda (x) (= (% x 7) 0)))", + "(collections.filter (list (list 1 2) (list 3) (list 1 2)) (lambda (x) (= x (list 1 2))))", + "(collections.reduce (collections.range 1000) + 0)", + "(collections.reduce (collections.range 10) (lambda (acc x) (cons x acc)) (list))", + "(collections.zip (collections.range 300) (collections.range 5 400))", + "(collections.flatten (list (collections.range 300) 1 (list) (collections.range 3)))", + "(collections.sort (collections.map (collections.range 300) (lambda (x) (% (* x 37) 101))))", + "(collections.sort (list \"pear\" \"apple\" \"fig\" 3 1.5 2))", + "(collections.reverse (collections.range 300))", + "(collections.unique (collections.map (collections.range 300) (lambda (x) (% x 17))))", + "(collections.unique (list (list 1 2) (list 1 2) (list 1 (list 2)) (list 1 (list 2))))", + "(collections.dict-keys (dict \"a\" 1 \"b\" 2))", + "(collections.dict-values (dict \"a\" 1 \"b\" 2))", + "(collections.dict-merge (dict \"a\" 1 \"b\" 2 \"a\" 9) (dict \"b\" 3 \"c\" 4 \"c\" 5))", + // The builtins that call a function take a DSL function too. + "(apply (lambda (a b) (+ a b)) (list 1 2))", + "(apply (lambda (x) (return (* x 2))) (list 21))", + "(apply collections.map (list (list 1 2) (lambda (x) (* x 3))))", + "(collections.map (list (list 1 2) (list 3 4 5)) (lambda (xs) (apply + xs)))", + "(begin (set o (dict \"__class__\" (dict \"inc\" (lambda (self x) (+ x 1))))) (send o \"inc\" " + "41))", +}; + +} // namespace + +TEST(StdlibSlicedTest, SameResultsHoweverTheWorkIsSliced) { + sliced_scheduler whole(0), one(1), three(3), some(64), host(1024); + for (const char *script : kSlicedScripts) { + SCOPED_TRACE(script); + const std::string expected = whole.run(script).first; + EXPECT_EQ(one.run(script).first, expected); + EXPECT_EQ(three.run(script).first, expected); + EXPECT_EQ(some.run(script).first, expected); + EXPECT_EQ(host.run(script).first, expected); + } + const auto r = [&](const char *script) { return three.run(script).first; }; + EXPECT_EQ(r("(string.split \"a,b,,c,\" \",\")"), "(\"a\" \"b\" \"\" \"c\" \"\")"); + EXPECT_EQ(r("(string.split \"one two three\")"), "(\"one\" \"two\" \"\" \"three\")"); + EXPECT_EQ(r("(string.split \"\" \",\")"), "(\"\")"); + EXPECT_EQ(r("(string.split \"caf\xC3\xA9 \xE2\x82\xAC\" \"\")"), + "(\"c\" \"a\" \"f\" \"\xC3\xA9\" \" \" \"\xE2\x82\xAC\")"); + EXPECT_EQ(r("(length (string.split (string.join (collections.map (collections.range 500) str) " + "\"::\") \"::\"))"), + "500"); + EXPECT_EQ(r("(string.replace \"aaaa\" \"aa\" \"b\")"), "bb"); + EXPECT_EQ(r("(bytes.decode (bytes.slice (string.encode \"hello, w\xC3\xB6rld\") 7))"), + "w\xC3\xB6rld"); + EXPECT_EQ(r("(collections.map (list (list 1 2) (list 3)) (lambda (xs) (collections.map xs " + "(lambda (x) (* 10 x)))))"), + "((10 20) (30))"); + EXPECT_EQ(r("(collections.map (list 1 -2 3) (lambda (x) (if (< x 0) (return 0) x)))"), "(1 0 3)"); + EXPECT_EQ(r("(collections.map (list 2 3) (lambda (n) (begin (set i 0) (while t (begin (set i " + "(+ i 1)) (if (= i n) (break (* i 10)) nil))))))"), + "(20 30)"); + EXPECT_EQ(r("(collect (generator (lambda () (yield (collections.map (list 1 2) (lambda (x) (+ " + "x 1)))))))"), + "((2 3))"); + EXPECT_EQ(r("(collections.filter (list (list 1 2) (list 3) (list 1 2)) (lambda (x) (= x (list " + "1 2))))"), + "((1 2) (1 2))"); + EXPECT_EQ(r("(collections.reduce (collections.range 1000) + 0)"), "499500"); + EXPECT_EQ(r("(collections.reduce (collections.range 10) (lambda (acc x) (cons x acc)) (list))"), + "(9 8 7 6 5 4 3 2 1 0)"); + EXPECT_EQ(r("(collections.sort (list 5 3.5 -1 2 10))"), "(-1 2 3.5 5 10)"); + EXPECT_EQ(r("(collections.sort (list \"pear\" \"apple\" \"fig\"))"), + "(\"apple\" \"fig\" \"pear\")"); + EXPECT_EQ(r("(collections.unique (list (list 1 2) (list 1 2) (list 1 (list 2)) (list 1 (list " + "2))))"), + "((1 2) (1 (2)))"); + EXPECT_EQ(r("(apply (lambda (a b) (+ a b)) (list 1 2))"), "3"); + EXPECT_EQ(r("(apply (lambda (x) (return (* x 2))) (list 21))"), "42"); + EXPECT_EQ(r("(apply collections.map (list (list 1 2) (lambda (x) (* x 3))))"), "(3 6)"); + EXPECT_EQ(r("(collections.map (list (list 1 2) (list 3 4 5)) (lambda (xs) (apply + xs)))"), + "(3 12)"); + EXPECT_EQ(r("(begin (set o (dict \"__class__\" (dict \"inc\" (lambda (self x) (+ x 1))))) " + "(send o \"inc\" 41))"), + "42"); + EXPECT_EQ(r("(collections.dict-merge (dict \"a\" 1 \"b\" 2 \"a\" 9) (dict \"b\" 3 \"c\" 4 " + "\"c\" 5))"), + "{\"a\": 1, \"b\": 3, \"a\": 9, \"c\": 5}"); +} + +TEST(StdlibSlicedTest, ACallSpreadsOverDataSteps) { + const std::string script = + "(begin (set s (string.join (collections.map (collections.range 5000) str) \",\"))" + " (length (string.split s \",\")))"; + sliced_scheduler whole(0), sliced(16); + const auto [r0, i0] = whole.run(script); + const auto [r1, i1] = sliced.run(script); + EXPECT_EQ(r0, "5000"); + EXPECT_EQ(r1, "5000"); + EXPECT_EQ(i1.eval_steps, i0.eval_steps); // the same evaluation... + EXPECT_GT(i1.data_steps, i0.data_steps + 3 * 5000 / 16); // ...its data work a slice a step + EXPECT_GE(i1.data_nodes, 3u * 5000); // map, join and split visit each + EXPECT_EQ(i1.eval_steps + i1.data_steps + i1.read_steps, i1.step_count); +} + +TEST(StdlibSlicedTest, MaxStepsStopsAHugeCallMidway) { + const std::string script = + "(length (collections.sort (collections.reverse (collections.range 20000))))"; + execute_options opts; + opts.max_steps = 200; + sliced_scheduler whole(0), sliced(16); + const auto [r0, i0] = whole.run(script, opts); // unsliced it fits in a few steps + EXPECT_EQ(r0, "20000"); + EXPECT_EQ(i0.status, process_status::terminated); + EXPECT_LT(i0.step_count, 20u); + const auto [r1, i1] = sliced.run(script, opts); + EXPECT_EQ(i1.status, process_status::killed); + EXPECT_LE(i1.step_count, 201u); + EXPECT_GT(i1.data_steps, 150u); +} + +TEST(StdlibSlicedTest, MaxTimeStopsAHugeCall) { + async_scheduler sched; + sched.set_data_defaults(async_scheduler::host_data_defaults()); + execute_options opts; + opts.max_time = 0.02; + const int pid = + sched.execute(std::string("(collections.unique (collections.range 20000))"), opts); + sched.sync_run(); + EXPECT_EQ(sched.get_process_info(pid)->status, process_status::killed); +} + +TEST(StdlibSlicedTest, OtherProcessesRunBetweenSlices) { + sliced_scheduler s(16); + const int slow = s.sched.execute( + std::string("(length (collections.sort (collections.reverse (collections.range 5000))))")); + while (s.sched.get_process_info(slow)->data_steps == 0) + s.sched.step(); + const int quick = s.sched.execute(std::string("(string.join (list \"a\" \"b\") \"-\")")); + while (s.sched.get_process_info(quick)->status != process_status::terminated) + s.sched.step(); + EXPECT_EQ(show(s.sched.get_result(quick)), "a-b"); + EXPECT_NE(s.sched.get_process_info(slow)->status, process_status::terminated); + s.sched.run(); + EXPECT_EQ(show(s.sched.get_result(slow)), "5000"); +} + +TEST(StdlibSlicedTest, TheAllocationCheckStillComesFirst) { + // A 1000-character string fits under 4 KiB; its 1000 one-character parts do not. Counting + // them takes a few one-node slices; the check then stops the call before it builds anything. + cvc::app app; + cvc::state &root = cvc::state::instance(app); + root("state_exec.defaults.data.step_nodes").value("1"); + root("state_exec.defaults.limits.max_alloc_bytes").value("4096"); + const std::string split = "(string.split \"" + std::string(1000, 'a') + "\" \"\")"; + const std::string err = thrown([&] { run_sync(split, {}, &root, "lim"); }); + EXPECT_NE(err.find("string.split"), std::string::npos) << err; + EXPECT_NE(err.find("allocation limit"), std::string::npos) << err; + // (a scheduler publishes what it resolved under its id, so a new id reads the new default) + root("state_exec.defaults.limits.max_alloc_bytes").value("1048576"); + EXPECT_EQ(run_sync("(length " + split + ")", {}, &root, "lim2"), "1000"); +} + +TEST(StdlibSlicedTest, MapFilterReduceCallDslFunctions) { + sliced_scheduler s(0); + // Each call is the process's own frames: the job resumes in a data step after each one. + const auto [r, info] = s.run("(collections.map (collections.range 100) (lambda (x) (+ x 1)))"); + EXPECT_EQ(r.substr(0, 9), "(1 2 3 4 "); + EXPECT_GE(info.data_steps, 100u); + EXPECT_GE(info.eval_steps, 300u); + EXPECT_EQ( + s.run("(begin (set k 3) (defun scale (x) (* k x)) (collections.map (list 1 2) scale))").first, + "(3 6)"); + EXPECT_EQ(s.run("(collections.map (list 1 2) (lambda (&rest xs) xs))").first, "((1) (2))"); + EXPECT_EQ(s.run("(collections.reduce (list \"a\" \"b\") (lambda (acc x) (str-concat acc x)) " + "\"\")") + .first, + "ab"); + EXPECT_NE(thrown([&] { s.run("(collections.map (list 1) (lambda (x y) x))"); }).find("closure"), + std::string::npos); + EXPECT_NE(thrown([&] { + s.run("(collections.map (list 1) 5)"); + }).find("collections.map: second argument must be a function"), + std::string::npos); + // The async scheduler (the app's exec scheduler type) the same. + std::string err; + EXPECT_EQ(run_async("(collections.filter (list 1 2 3 4) (lambda (x) (= (% x 2) 0)))", &err), + "(2 4)"); + EXPECT_EQ(err, ""); +} + +TEST(StdlibSlicedTest, ADslFunctionNeedsARunningProgram) { + // Called directly (no evaluator to run its frames), a job can call a native function only. + auto cls = std::make_shared(); + cls->params = {symbol{"x"}}; + cls->body = {value_t{symbol{"x"}}}; + cls->env_snapshot = builtins::make_default_environment(); + const value_t one_item = make_list({value_t(int64_t(-1))}); + EXPECT_NE(thrown([&] { + call("collections.map", {one_item, value_t{cls}}); + }).find("needs a running program"), + std::string::npos); + const native_fn *abs = stdlib_registry::standard().lookup_qualified("math.abs"); + ASSERT_NE(abs, nullptr); + EXPECT_EQ(show(call("collections.map", {one_item, value_t{*abs}})), "(1)"); + + // The same for apply, and for call_function() from any native function. + const std::vector apply_args = {value_t{cls}, one_item}; + EXPECT_NE(thrown([&] { + builtin_apply(std::span(apply_args.data(), apply_args.size())); + }).find("apply: calling a DSL function needs a running program"), + std::string::npos); + const value_t arg(int64_t(7)); + EXPECT_EQ(show(call_function(value_t{*abs}, std::span(&arg, 1), "test")), "7"); + EXPECT_NE(thrown([&] { + call_function(value_t{int64_t(1)}, std::span(&arg, 1), "test"); + }).find("test: not a function"), + std::string::npos); +}