diff --git a/bindings/pycvc/README.md b/bindings/pycvc/README.md index 3624fa02b..36e1ffc10 100644 --- a/bindings/pycvc/README.md +++ b/bindings/pycvc/README.md @@ -203,9 +203,16 @@ ex.parser_defaults() # == host_parser_defaults( pycvc.state_set(app, "state_exec.defaults.parser.max_depth", "64") # tree overrides root = pycvc.State.instance(app) # the tree root, as in C++ pycvc.parse_options_from_state(root, "", ex.parser_defaults()) # what Exec uses + +# Deep-data settings ("Deep data on the executor"), the same way +pycvc.data_options() # step_nodes 0 +pycvc.async_scheduler.host_data_defaults() # step_nodes 1024 +ex.data_defaults() # == host_data_defaults() +pycvc.data_options_from_state(root, "", ex.data_defaults()) # what Exec uses ``` `Exec` starts from the same defaults as the app's exec scheduler (Ariadne), and -the same `state_exec.defaults.parser.*` keys override both. A syntax error +the same `state_exec.defaults.parser.*` and `state_exec.defaults.data.*` keys +override both. A syntax error raises a `RuntimeError` with the parse error's message. Contract test: `test_pycvc_exec.py`. diff --git a/bindings/pycvc/pycvc.i b/bindings/pycvc/pycvc.i index 50fc32b62..f0db6d748 100644 --- a/bindings/pycvc/pycvc.i +++ b/bindings/pycvc/pycvc.i @@ -1445,8 +1445,11 @@ class AsyncStateObserver(state_observer): // async_scheduler.h (value_t, the parser, the scheduler itself) is not wrapped // yet, so async_scheduler has no Python constructor. Exec's parser_defaults / // set_parser_defaults are in pycvc_exec.h. +// The deep-data settings (data_options, data_options_from_state, "Deep data on the +// executor") are wrapped the same way. %{ #include +#include #include %} namespace cvc { @@ -1455,14 +1458,21 @@ struct parse_options { size_t max_depth; // lists and quotes that may nest (default 256) size_t step_tokens; // 0: parse at submit; N: read N tokens per process step }; +struct data_options { + size_t step_nodes; // 0: deep-data work in the step that needs it; N: N nodes per step +}; parse_options parse_options_from_state(cvc::state &root, const std::string &scheduler_id = std::string(), const parse_options &base = parse_options()); +data_options data_options_from_state(cvc::state &root, + const std::string &scheduler_id = std::string(), + const data_options &base = data_options()); %nodefaultctor async_scheduler; %nodefaultdtor async_scheduler; class async_scheduler { public: static parse_options host_parser_defaults(); + static data_options host_data_defaults(); }; } } diff --git a/bindings/pycvc/pycvc_exec.cpp b/bindings/pycvc/pycvc_exec.cpp index dd3893939..474842084 100644 --- a/bindings/pycvc/pycvc_exec.cpp +++ b/bindings/pycvc/pycvc_exec.cpp @@ -379,9 +379,11 @@ Exec::Exec(const std::shared_ptr &app) : impl_(std::make_sharedcomputePool(); impl_->sched.set_watch_root(impl_->root); // The same state_exec defaults as the app's exec scheduler, so a program behaves the same from - // Python as from Ariadne: scripts are read inside the process (host_parser_defaults), and the - // app tree's state_exec.defaults.parser.* keys override them for both. + // Python as from Ariadne: scripts are read inside the process (host_parser_defaults) and deep + // data is handled a slice per step (host_data_defaults), and the app tree's + // state_exec.defaults.parser.* / data.* keys override them for both. impl_->sched.set_parser_defaults(async_scheduler::host_parser_defaults()); + impl_->sched.set_data_defaults(async_scheduler::host_data_defaults()); impl_->host_proc = make_process(); impl_->host_proc->pid = 0; @@ -656,4 +658,10 @@ void Exec::set_parser_defaults(const cvc::state_exec::parse_options &defaults) { impl_->sched.set_parser_defaults(defaults); } +cvc::state_exec::data_options Exec::data_defaults() const { return impl_->sched.data_defaults(); } + +void Exec::set_data_defaults(const cvc::state_exec::data_options &defaults) { + impl_->sched.set_data_defaults(defaults); +} + } // namespace pycvc diff --git a/bindings/pycvc/pycvc_exec.h b/bindings/pycvc/pycvc_exec.h index b774ac2c4..27da26732 100644 --- a/bindings/pycvc/pycvc_exec.h +++ b/bindings/pycvc/pycvc_exec.h @@ -7,7 +7,8 @@ // registered here becomes callable from DSL source like any builtin. #pragma once -#include // parse_options +#include // data_options +#include // parse_options #include #include @@ -94,6 +95,11 @@ class Exec { cvc::state_exec::parse_options parser_defaults() const; void set_parser_defaults(const cvc::state_exec::parse_options &defaults); + // The deep-data settings, likewise: async_scheduler::data_defaults() / set_data_defaults(), + // starting as async_scheduler::host_data_defaults(). + cvc::state_exec::data_options data_defaults() const; + void set_data_defaults(const cvc::state_exec::data_options &defaults); + private: struct ExecImpl; std::shared_ptr impl_; diff --git a/bindings/pycvc/test_pycvc_exec.py b/bindings/pycvc/test_pycvc_exec.py index 3bf8f1eee..244fbb6ef 100644 --- a/bindings/pycvc/test_pycvc_exec.py +++ b/bindings/pycvc/test_pycvc_exec.py @@ -129,6 +129,63 @@ def test_set_parser_defaults_changes_how_exec_parses(): assert ex.run("(+ 1 (+ 2 3))") == "6" +# ── state_exec deep-data settings (docs/STATE_EXEC_GRAMMAR.md, "Deep data on +# the executor"), under the C++ names ──────────────────────────────────── + + +def test_data_options_is_the_cpp_struct(): + o = pycvc.data_options() + assert o.step_nodes == 0 # C++ default: deep-data work runs whole + o.step_nodes = 64 + assert o.step_nodes == 64 + + +def test_exec_starts_from_the_host_data_defaults(): + # The same defaults as the app's exec scheduler: + # async_scheduler::host_data_defaults(). + host = pycvc.async_scheduler.host_data_defaults() + assert host.step_nodes == 1024 + ex = pycvc.Exec(pycvc.make_app()) + assert ex.data_defaults().step_nodes == host.step_nodes + + +def test_data_options_from_state_resolves_like_cpp(): + # The real C++ function: it takes the tree root, a State, as in C++. + app = pycvc.make_app() + root = pycvc.State.instance(app) + assert pycvc.data_options_from_state(root).step_nodes == 0 + ex = pycvc.Exec(app) + assert pycvc.data_options_from_state(root, "", ex.data_defaults()).step_nodes == 1024 + root.child("state_exec.defaults.data.step_nodes").value("16") + root.child("state_exec.schedulers.s1.data.step_nodes").value("8") + assert pycvc.data_options_from_state(root).step_nodes == 16 + assert pycvc.data_options_from_state(root, "s1").step_nodes == 8 + root.child("state_exec.defaults.data.step_nodes").value("lots") # bad: skipped + assert pycvc.data_options_from_state(root).step_nodes == 0 + o = pycvc.data_options() + o.step_nodes = 0 + ex.set_data_defaults(o) + assert ex.data_defaults().step_nodes == 0 + + +def test_deep_data_gives_the_same_results_however_it_is_sliced(): + # Compare and print values nested 3000 deep: whole in one step, 1024 nodes a step (the host + # default) or 4 nodes a step. + app = pycvc.make_app() + ex = pycvc.Exec(app) + prog = ( + "(begin (set a 1) (set b 1) (set i 0)" + " (while (< i 3000) (begin (set a (list a)) (set b (list b)) (set i (+ i 1))))" + " (list (= a b) (length (str a))))" + ) + sliced = ex.run(prog) + pycvc.state_set(app, "state_exec.defaults.data.step_nodes", "0") + whole = ex.run(prog) + pycvc.state_set(app, "state_exec.defaults.data.step_nodes", "4") + assert ex.run(prog) == whole == sliced + assert whole.startswith("(#t ") + + # ── DSL <-> state <-> Python round trips (one shared app) ─────────────── diff --git a/docs/STATE_EXEC_GRAMMAR.md b/docs/STATE_EXEC_GRAMMAR.md index c6045b560..00df8b858 100644 --- a/docs/STATE_EXEC_GRAMMAR.md +++ b/docs/STATE_EXEC_GRAMMAR.md @@ -208,6 +208,64 @@ scheduler can parse a script the same way it evaluates one: a slice per step. Wi - **Snapshots.** A process snapshotted while reading stores its source and parser settings. After a restore it reads the script again from the start. +### Deep data on the executor + +Walking a nested value no longer recurses on the native stack: freeing, printing (`to_string`), +comparing (`values_equal`), copying (`state-data-get`), macro expansion and the snapshot codec +keep their work on the heap. That work is still proportional to the value's size, so a scheduler +can run it the way it reads a script: a slice per step, counted against the process's budget. + +The setting is `data.step_nodes`, resolved like the parser's (`state_exec.schedulers..data.step_nodes`, +then `state_exec.defaults.data.step_nodes`, then the scheduler's built-in default; +`data_options_from_state()`): + +| Key | Meaning | Default | +|---|---|---| +| `step_nodes` | 0: deep-data work runs whole, inside the step that needs it. N: a step does at most N nodes of it; the rest runs N per step in the steps after | 0; **1024** for the app's exec scheduler and pycvc's `Exec` | + +A node is an element visited: a list element or a dict entry walked, or an element freed. With +`step_nodes` = N: + +- **Native calls.** `=`, `!=`, `str`, `str-concat`, `print` and `state-data-get` hand their walk + to the evaluator (`run_data_job`). It does one slice inside the call's step; if the walk isn't + done, the call's frame waits and the following steps are **data steps** that each run N more + nodes. The step that finishes it completes the call. A walk under N nodes finishes in the + call's own step, so ordinary programs take no extra steps. +- **Macro expansion** works the same way, before the expansion is evaluated. +- **Freeing.** Last owners dropped during a step (a list, dict, closure, generator or scope) + go to the process's release backlog instead of being freed on the spot. The step frees up to + what is left of its N nodes at its end. If more is left, the next steps are data steps that + only free, and evaluation resumes once the backlog is empty. +- **Budgets.** Data steps are process steps. They count against `max_steps` and `max_time`, and + 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. +- **Snapshots** keep `step_nodes` but not a walk in flight: after a restore, the waiting call + or macro expansion runs again. +- **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`. + +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. + +### Budget by kind + +Each process's steps are split by what they were spent on (`step_kind`): + +- **read**: reading the script (`parser.step_tokens`); +- **data**: deep-data steps (`data.step_nodes`); +- **eval**: evaluating code. + +`process_info` carries `read_steps`, `eval_steps`, `data_steps` (they add up to `step_count`), +`data_nodes` (deep-data nodes, whichever steps did them), and `read_time`, `eval_time`, +`data_time`. The times are the seconds of the steps a scheduler ran, charged to each step's kind. +`inspect` reports them for any pid, and `(budget-usage)` reports them, with `steps`, for the +running process. + ## Data | Production | Value (`value_t` alternative) | diff --git a/inc/cvc/state/state_exec/async_scheduler.h b/inc/cvc/state/state_exec/async_scheduler.h index 8af85f86f..9c2231a93 100644 --- a/inc/cvc/state/state_exec/async_scheduler.h +++ b/inc/cvc/state/state_exec/async_scheduler.h @@ -182,6 +182,30 @@ class async_scheduler : public scheduler_base { return defaults; } + /// This scheduler's built-in deep-data settings: what a submitted process runs with when the + /// watch root's `state_exec.defaults.data.*` keys don't say otherwise. data_options{} (no + /// slicing) unless the host sets it. + void set_data_defaults(const data_options &defaults) { data_defaults_ = defaults; } + const data_options &data_defaults() const { return data_defaults_; } + + /// The deep-data settings a process submitted now gets: the watch root's keys over + /// data_defaults(). Read on every submit, like the parser settings. + data_options data_settings() const { + return watch_root_ ? data_options_from_state(*watch_root_, {}, data_defaults_) : data_defaults_; + } + + /// The deep-data defaults libcvc's own hosts give their async schedulers (app::exec_scheduler() + /// and pycvc's Exec), next to host_parser_defaults(): deep-data work runs 1024 nodes per step + /// (docs/STATE_EXEC_GRAMMAR.md, "Deep data on the executor"). At the measured ~45 ns a node + /// to free, ~70 to print, ~250 to compare and ~380 to copy, a slice costs 0.05-0.4 ms, so a huge + /// value is spread over steps instead of stalling a drain, while the data of a normal program + /// (far under 1024 nodes a step) costs it no extra steps. + static data_options host_data_defaults() { + data_options defaults; + defaults.step_nodes = 1024; + return defaults; + } + private: scheduling_policy policy_; int next_pid_ = 1; @@ -200,6 +224,7 @@ class async_scheduler : public scheduler_base { memory_tracker mem_tracker_; cvc::state *watch_root_ = nullptr; parse_options parser_defaults_; + data_options data_defaults_; std::unordered_map processes_; diff --git a/inc/cvc/state/state_exec/data_job.h b/inc/cvc/state/state_exec/data_job.h new file mode 100644 index 000000000..52d41a141 --- /dev/null +++ b/inc/cvc/state/state_exec/data_job.h @@ -0,0 +1,164 @@ +/** + * @file data_job.h + * @brief Deep-data work (comparing, printing, copying, expanding nested values) in slices. + * + * A walk over a nested value costs one unit per node it visits. A data_job keeps its walk on the + * heap, so it can stop after any number of nodes and go on later. A scheduler runs a process's + * deep-data work a slice at a time (data.step_nodes on the state tree), as it reads a script a + * slice at a time (parser.step_tokens): the work counts against the process's max_steps and + * max_time, and other processes run between the slices. + */ +#ifndef CVC_STATE_EXEC_DATA_JOB_H +#define CVC_STATE_EXEC_DATA_JOB_H + +#include +#include +#include +#include +#include +#include +#include + +namespace cvc { +class state; +} + +namespace cvc::state_exec { + +struct evaluation_stats; + +/// 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 +/// result then reflects each part as it was when visited. +class data_job { +public: + virtual ~data_job() = default; + + /// Visit at most `nodes` more nodes; 0 visits all that remain. True once the walk is done. + virtual bool step(std::size_t nodes) = 0; + + /// The walk's result, once step() has returned true. Call it once. + virtual value_t result() = 0; + + /// Nodes visited so far. + std::size_t nodes() const { return nodes_; } + +protected: + std::size_t nodes_ = 0; +}; + +/// Whether walking `v` visits any nodes: it is a non-empty list or dict. +bool is_compound(const value_t &v); + +/// values_equal(a, b), as a bool value. +std::unique_ptr make_equal_job(value_t a, value_t b); + +/// to_string(v), as a string value. +std::unique_ptr make_print_job(value_t v); + +/// `parts` printed one after another with `sep` between them, as a string value: a string as its +/// raw text, anything else as to_string() prints it. The same string may be passed many times, so +/// the result can outgrow every part: each append is checked against the allocation limit +/// (eval_alloc_check), naming `who` (a string literal: the builtin) when it is over. +std::unique_ptr make_concat_job(std::vector parts, std::string sep, + const char *who); + +/// A private copy of v's lists and dicts, keeping their sharing (a node reached twice is copied +/// once; a cycle is copied as a cycle). Anything else is shared, not copied. +std::unique_ptr make_copy_job(value_t v); + +/// A macro template with its parameters replaced: each symbol named in `subst` becomes its +/// value; lists are copied, everything else is kept. +std::unique_ptr make_substitute_job(value_t tmpl, + std::unordered_map subst); + +/// Run a job to completion and return its result. +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 +/// on the state tree under `data.*`; data_options_from_state() reads them from there. +struct data_options { + /// How many nodes of deep-data work a scheduler step may do. 0 does all of it in the step that + /// needs it. N > 0 does up to N there; what is left runs N per step in the steps after it + /// (the process waits, and those steps count against its max_steps and max_time). Freeing + /// counts too: up to N owners a step dropped are freed at its end, the rest in steps after. + /// State-tree key: `data.step_nodes`. + std::size_t step_nodes = 0; +}; + +/// Resolve data settings from a state tree, per key, like parse_options_from_state(): +/// `state_exec.schedulers..data.`, then `state_exec.defaults.data.`, then +/// `base`. A value that is missing, empty or malformed is skipped. +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. +value_t run_data_job(std::unique_ptr job, std::function then = {}); + +/// 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); + +namespace detail { + +/// A native call's deep-data job left to run after the call returned (see run_data_job). +struct deferred_data_job { + std::unique_ptr job; + std::function then; +}; + +/// The data side of one stackless evaluator step, armed for the step's duration (nested evaluator +/// steps arm their own and restore this one). It counts the step's deep-data nodes into `stats` +/// and, when the step slices, routes the owners it drops into `backlog` (see release_backlog). +class data_step_scope { +public: + data_step_scope(std::size_t step_nodes, evaluation_stats &stats, release_backlog *backlog); + ~data_step_scope(); + data_step_scope(const data_step_scope &) = delete; + data_step_scope &operator=(const data_step_scope &) = delete; + + /// The slice: how many nodes the step may visit (0 = no slicing). + std::size_t step_nodes() const { return step_nodes_; } + /// Nodes counted in this step so far. + std::size_t nodes() const { return nodes_; } + /// What is left of the slice (never 0 while slicing, so a job always advances). + std::size_t remaining() const; + + /// Let run_data_job() defer a job while `allowed` (around the native call being applied). + void allow_defer(bool allowed) { may_defer_ = allowed; } + /// The job a native call deferred, if any. + std::unique_ptr take_deferred() { return std::move(deferred_); } + + /// Free backlog owners with what is left of the slice (all of them when not slicing). + void drain_backlog(); + + /// The scope armed on this thread, if any. + static data_step_scope *current(); + +private: + friend value_t cvc::state_exec::run_data_job(std::unique_ptr, + std::function); + friend void cvc::state_exec::count_data_nodes(std::size_t); + + std::size_t step_nodes_; + evaluation_stats &stats_; + release_backlog *backlog_; + std::size_t nodes_ = 0; + bool may_defer_ = false; + std::unique_ptr deferred_; + data_step_scope *outer_; + backlog_scope routing_; +}; + +} // namespace detail + +} // namespace cvc::state_exec + +#endif // CVC_STATE_EXEC_DATA_JOB_H diff --git a/inc/cvc/state/state_exec/evaluator.h b/inc/cvc/state/state_exec/evaluator.h index 16e2af8c4..b10171e9f 100644 --- a/inc/cvc/state/state_exec/evaluator.h +++ b/inc/cvc/state/state_exec/evaluator.h @@ -43,18 +43,26 @@ class return_exception : public std::exception { const char *what() const noexcept override { return "return"; } }; +/// What a step was spent on: evaluating code, reading the script a slice at a time +/// (parser.step_tokens), or deep-data work a slice at a time (data.step_nodes: comparing, +/// printing, copying or expanding a nested value, or freeing one). +enum class step_kind { eval = 0, read = 1, data = 2 }; + /// Thread-safe evaluation statistics. struct evaluation_stats { evaluation_stats() = default; // Move-only: transfer ownership of internals evaluation_stats(evaluation_stats &&o) noexcept - : start_time(o.start_time), end_time(o.end_time), - step_count(o.step_count.load(std::memory_order_relaxed)), complete(o.complete) {} + : start_time(o.start_time), end_time(o.end_time), complete(o.complete), + last_kind(o.last_kind) { + copy_counts(o); + } evaluation_stats &operator=(evaluation_stats &&o) noexcept { start_time = o.start_time; end_time = o.end_time; - step_count.store(o.step_count.load(std::memory_order_relaxed), std::memory_order_relaxed); complete = o.complete; + last_kind = o.last_kind; + copy_counts(o); return *this; } evaluation_stats(const evaluation_stats &) = delete; @@ -64,6 +72,11 @@ struct evaluation_stats { std::lock_guard lk(mu); start_time = std::chrono::steady_clock::now(); step_count.store(0, std::memory_order_relaxed); + for (std::size_t k = 0; k < kKinds; ++k) { + kind_steps[k].store(0, std::memory_order_relaxed); + kind_ns[k].store(0, std::memory_order_relaxed); + } + data_node_count.store(0, std::memory_order_relaxed); complete = false; } void mark_complete() { @@ -71,8 +84,42 @@ struct evaluation_stats { end_time = std::chrono::steady_clock::now(); complete = true; } - void increment_step() { step_count.fetch_add(1, std::memory_order_relaxed); } + /// Count a step, as spent on `kind`. + void increment_step(step_kind kind = step_kind::eval) { + step_count.fetch_add(1, std::memory_order_relaxed); + if (kind != step_kind::eval) // eval steps are the rest, so the common step costs no more + kind_steps[static_cast(kind)].fetch_add(1, std::memory_order_relaxed); + last_kind = kind; + } uint64_t get_step_count() const { return step_count.load(std::memory_order_relaxed); } + /// Steps spent on `kind`. + uint64_t steps(step_kind kind) const { + if (kind != step_kind::eval) + return kind_steps[static_cast(kind)].load(std::memory_order_relaxed); + const uint64_t other = steps(step_kind::read) + steps(step_kind::data); + const uint64_t all = get_step_count(); + return all > other ? all - other : 0; + } + /// The kind of the last step counted. + step_kind last_step_kind() const { return last_kind; } + + /// Count nodes of deep-data work (elements walked, owners freed), in whatever step did it. + void add_data_nodes(uint64_t nodes) { + data_node_count.fetch_add(nodes, std::memory_order_relaxed); + } + uint64_t data_nodes() const { return data_node_count.load(std::memory_order_relaxed); } + + /// A scheduler that times its steps charges each step's seconds here, to the step's kind. + void add_step_time(double seconds) { + if (seconds > 0.0) + kind_ns[static_cast(last_kind)].fetch_add(static_cast(seconds * 1e9), + std::memory_order_relaxed); + } + /// Seconds of scheduler-timed steps spent on `kind`. + double step_time(step_kind kind) const { + return 1e-9 * static_cast( + kind_ns[static_cast(kind)].load(std::memory_order_relaxed)); + } bool is_complete() const { std::lock_guard lk(mu); return complete; @@ -84,11 +131,27 @@ struct evaluation_stats { } private: + static constexpr std::size_t kKinds = 3; + void copy_counts(const evaluation_stats &o) { + step_count.store(o.step_count.load(std::memory_order_relaxed), std::memory_order_relaxed); + for (std::size_t k = 0; k < kKinds; ++k) { + kind_steps[k].store(o.kind_steps[k].load(std::memory_order_relaxed), + std::memory_order_relaxed); + kind_ns[k].store(o.kind_ns[k].load(std::memory_order_relaxed), std::memory_order_relaxed); + } + data_node_count.store(o.data_node_count.load(std::memory_order_relaxed), + std::memory_order_relaxed); + } + mutable std::mutex mu; std::chrono::steady_clock::time_point start_time{}; std::chrono::steady_clock::time_point end_time{}; std::atomic step_count{0}; + std::atomic kind_steps[kKinds] = {}; + std::atomic kind_ns[kKinds] = {}; + std::atomic data_node_count{0}; bool complete{false}; + step_kind last_kind = step_kind::eval; }; /// Synchronous recursive evaluator (Tier 2 — local contexts only). diff --git a/inc/cvc/state/state_exec/parser.h b/inc/cvc/state/state_exec/parser.h index 5267c0c3d..617adfb33 100644 --- a/inc/cvc/state/state_exec/parser.h +++ b/inc/cvc/state/state_exec/parser.h @@ -42,9 +42,10 @@ class parse_error : public std::runtime_error { /// state tree under `parser.*`; parse_options_from_state() reads them from there. struct parse_options { /// How many lists and quotes may nest (`'x` reads as `(quote x)`, one level); deeper input is - /// a parse_error. The parser itself is stackless, but code downstream still walks values - /// recursively (destroying a nested list, the snapshot codec, to_string), so this keeps parsed - /// data within what they can handle. State-tree key: `parser.max_depth`. + /// a parse_error. The parser is stackless, and so are freeing, printing, comparing, copying, + /// macro-expanding and snapshotting a value; what still walks values recursively (the Ariadne + /// loader's channel lint, which stops at 256 levels, and pycvc's value conversion) is what this + /// keeps parsed data within. State-tree key: `parser.max_depth`. std::size_t max_depth = 256; /// How a scheduler parses a submitted script. 0 parses all of it when it is submitted, so diff --git a/inc/cvc/state/state_exec/scheduler.h b/inc/cvc/state/state_exec/scheduler.h index 0160018a3..0401c6245 100644 --- a/inc/cvc/state/state_exec/scheduler.h +++ b/inc/cvc/state/state_exec/scheduler.h @@ -181,8 +181,9 @@ class scheduler : public scheduler_base { /// per-scheduler subtree so they are visible in the state tree. /// Requires set_watch_root() and (for per-scheduler) set_id() first. /// - /// Settings: `max_pending_messages`, and the parser's `parser.*` - /// (parse_options_from_state), which apply to every script submitted + /// Settings: `max_pending_messages`, the parser's `parser.*` + /// (parse_options_from_state) and the deep-data `data.*` + /// (data_options_from_state), which apply to every process submitted /// afterwards. void load_settings(); @@ -190,6 +191,10 @@ class scheduler : public scheduler_base { /// load_settings() runs). const parse_options &parser_settings() const { return parse_options_; } + /// The deep-data settings execute() gives processes (defaults until + /// load_settings() runs). + const data_options &data_settings() const { return data_options_; } + /// Well-known state tree paths for scheduler configuration. static constexpr const char *settings_root = "state_exec"; static constexpr const char *defaults_prefix = "state_exec.defaults"; @@ -204,6 +209,7 @@ class scheduler : public scheduler_base { uint64_t max_alloc_bytes_ = kDefaultMaxAllocBytes; scheduling_policy policy_; parse_options parse_options_; + data_options data_options_; int next_pid_ = 1; int rr_index_ = 0; bool running_ = false; diff --git a/inc/cvc/state/state_exec/scheduler_base.h b/inc/cvc/state/state_exec/scheduler_base.h index 179d6c292..f17e6371a 100644 --- a/inc/cvc/state/state_exec/scheduler_base.h +++ b/inc/cvc/state/state_exec/scheduler_base.h @@ -72,6 +72,17 @@ struct process_info { std::string uid; std::string gid; uint64_t step_count; + // What the steps were spent on (step_kind): reading the script a slice at a time + // (parser.step_tokens), deep-data work a slice at a time (data.step_nodes), evaluating the + // rest. data_nodes counts the deep-data work in nodes, whichever steps did it; the times are + // seconds of scheduler-timed steps. + uint64_t read_steps = 0; + uint64_t eval_steps = 0; + uint64_t data_steps = 0; + uint64_t data_nodes = 0; + double read_time = 0.0; + double eval_time = 0.0; + double data_time = 0.0; double elapsed_time; uint64_t current_memory; uint64_t peak_memory; @@ -84,6 +95,17 @@ struct process_info { int parent_pid; }; +/// Fill `info`'s budget-by-kind fields from a process's evaluation stats. +inline void fill_budget_usage(process_info &info, const evaluation_stats &stats) { + info.read_steps = stats.steps(step_kind::read); + info.eval_steps = stats.steps(step_kind::eval); + info.data_steps = stats.steps(step_kind::data); + info.data_nodes = stats.data_nodes(); + info.read_time = stats.step_time(step_kind::read); + info.eval_time = stats.step_time(step_kind::eval); + info.data_time = stats.step_time(step_kind::data); +} + /// Aggregate scheduler statistics. struct scheduler_stats { int total_processes = 0; diff --git a/inc/cvc/state/state_exec/stackless_evaluator.h b/inc/cvc/state/state_exec/stackless_evaluator.h index 10902797c..bcd88d470 100644 --- a/inc/cvc/state/state_exec/stackless_evaluator.h +++ b/inc/cvc/state/state_exec/stackless_evaluator.h @@ -14,6 +14,7 @@ #include #include #include +#include #include #include // parse_options #include @@ -64,6 +65,16 @@ 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) or the expansion of a macro call (its frame waits to be +/// re-initialized with it). +struct pending_data_work { + enum class kind { native_call, macro_expansion }; + std::unique_ptr job; + std::function then; // native_call: turns the job's result into the call's + kind what = kind::native_call; +}; + /// Serializable evaluator state — the entire execution snapshot. struct evaluator_state { std::vector stack; @@ -77,6 +88,17 @@ struct evaluator_state { /// Set while the script is still being read, a slice per step (parse_options::step_tokens); /// step() reads instead of evaluating until the whole script is read. std::shared_ptr reading; + /// How many nodes of deep-data work a step may do (data_options::step_nodes; a scheduler sets + /// 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. + 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 + /// while any are left. + std::shared_ptr backlog; }; /// Synchronous stackless evaluator (Tier 2 — local contexts only). @@ -147,6 +169,12 @@ class stackless_evaluator { evaluator_state create_program_state(std::vector forms, environment_ptr env); // One step of a state that is still reading its script (evaluator_state::reading). bool step_read(evaluator_state &state); + // One step of deep-data work left unfinished (evaluator_state::data_work). + bool step_data(evaluator_state &state); + // Re-initialize the top frame to evaluate a macro call's expansion. + void begin_expansion(evaluator_state &state, value_t expansion); + // One evaluation step of the top frame. + void step_frame(evaluator_state &state); // Frame management void pop_frame(evaluator_state &state, value_t result); @@ -175,7 +203,6 @@ class stackless_evaluator { void call_closure(evaluator_state &state, eval_frame &frame, const closure_ptr &cls, const std::vector &args); void handle_defclass(evaluator_state &state, eval_frame &frame, const std::vector &args); - value_t substitute(const value_t &tmpl, const std::unordered_map &subst); environment_ptr global_env_; std::shared_ptr> diff --git a/inc/cvc/state/state_exec/types.h b/inc/cvc/state/state_exec/types.h index b7e784043..4a2e76e9d 100644 --- a/inc/cvc/state/state_exec/types.h +++ b/inc/cvc/state/state_exec/types.h @@ -77,16 +77,59 @@ namespace detail { /// plain shared_ptr destructors takes a native stack frame per level and overflows the stack. /// Here the first release on a thread runs a loop: owners dropped while it frees something /// (that thing's elements, bindings, enclosing scope) are queued and freed one at a time by the -/// loop, so the stack stays a few frames deep whatever the depth. -void release_nested(std::shared_ptr owner) noexcept; +/// loop, so the stack stays a few frames deep whatever the depth. `nodes` is what freeing it +/// costs as deep-data work: 1, plus a list's or dict's elements. +void release_nested(std::shared_ptr owner, std::size_t nodes = 1) noexcept; /// Drop `owner` through release_nested when it is the last one; dropping any other owner only /// decrements a count. (Should another thread drop the last but one meanwhile, this drop frees /// the storage inline, and its elements start releases of their own: a few frames, still.) template void drop_owner(std::shared_ptr &owner) noexcept { - if (owner && owner.use_count() == 1) - release_nested(std::move(owner)); + if (!owner || owner.use_count() != 1) + return; + std::size_t nodes = 1; + if constexpr (std::is_same_v> || + std::is_same_v>>) + nodes += owner->size(); + release_nested(std::move(owner), nodes); } + +/// Last owners dropped while a slicing process step runs (data.step_nodes > 0) and not freed +/// yet. While a backlog_scope routes drops here, release_nested() queues a last owner instead of +/// freeing it, and the step frees them a slice at a time: freeing a huge value then counts +/// against the process's budget and spreads over its steps instead of stalling one. (A single +/// owner is freed whole: a flat list of a million numbers is one free.) +class release_backlog { +public: + release_backlog() = default; + ~release_backlog(); // frees what is left, iteratively like any release + release_backlog(const release_backlog &) = delete; + release_backlog &operator=(const release_backlog &) = delete; + + bool empty() const { return owners_.empty(); } + + /// Free owners until their nodes reach `budget` (0: until none are left). Owners dropped by + /// that freeing join the backlog. Returns the nodes freed. + std::size_t drain(std::size_t budget); + +private: + friend void release_nested(std::shared_ptr, std::size_t) noexcept; + std::vector, std::size_t>> owners_; +}; + +/// RAII: while alive, last owners this thread drops join `backlog` instead of being freed on the +/// spot. A null backlog changes nothing (an outer scope, if any, keeps routing). Scopes nest. +class backlog_scope { +public: + explicit backlog_scope(release_backlog *backlog); + ~backlog_scope(); + backlog_scope(const backlog_scope &) = delete; + backlog_scope &operator=(const backlog_scope &) = delete; + +private: + release_backlog *outer_; + bool armed_; +}; } // namespace detail /// The DSL value variant. All runtime values are represented as value_t. diff --git a/src/cvc/CMakeLists.txt b/src/cvc/CMakeLists.txt index 264aa30bf..8c440805a 100644 --- a/src/cvc/CMakeLists.txt +++ b/src/cvc/CMakeLists.txt @@ -87,6 +87,7 @@ set(INCLUDE_FILES ../../inc/cvc/state/state_exec/async_evaluator.h ../../inc/cvc/state/state_exec/async_stackless_evaluator.h ../../inc/cvc/state/state_exec/state_value_codec.h + ../../inc/cvc/state/state_exec/data_job.h ../../inc/cvc/state/state_exec/process.h ../../inc/cvc/state/state_exec/memory_tracker.h ../../inc/cvc/state/state_exec/scheduler_base.h @@ -209,6 +210,7 @@ set(SOURCE_FILES state/state_exec/async_evaluator.cpp state/state_exec/async_stackless_evaluator.cpp state/state_exec/state_value_codec.cpp + state/state_exec/data_job.cpp state/state_exec/process.cpp state/state_exec/memory_tracker.cpp state/state_exec/scheduler.cpp diff --git a/src/cvc/core/app.cpp b/src/cvc/core/app.cpp index 5af747911..5e4cdde77 100644 --- a/src/cvc/core/app.cpp +++ b/src/cvc/core/app.cpp @@ -1025,6 +1025,9 @@ cvc::state_exec::async_scheduler &app::exec_scheduler() { // Read submitted scripts inside their processes, like pycvc's Exec (see host_parser_defaults // for why 1024 tokens a step). state_exec.defaults.parser.* on this tree overrides it. _execScheduler->set_parser_defaults(cvc::state_exec::async_scheduler::host_parser_defaults()); + // Deep-data work a slice per step too (host_data_defaults); state_exec.defaults.data.* + // overrides it. + _execScheduler->set_data_defaults(cvc::state_exec::async_scheduler::host_data_defaults()); } return *_execScheduler; } diff --git a/src/cvc/state/state_exec/async_scheduler.cpp b/src/cvc/state/state_exec/async_scheduler.cpp index 46b210f24..578f40fec 100644 --- a/src/cvc/state/state_exec/async_scheduler.cpp +++ b/src/cvc/state/state_exec/async_scheduler.cpp @@ -46,6 +46,7 @@ int async_scheduler::execute(const std::string &script, const execute_options &o const parse_options popts = watch_root_ ? parse_options_from_state(*watch_root_, {}, parser_defaults_) : parser_defaults_; proc.state = evaluator_.create_state(script, opts.env, popts); + proc.state.data_step_nodes = data_settings().step_nodes; proc.state.stats.start(); processes_.emplace(pid, std::make_shared(std::move(proc))); return pid; @@ -75,6 +76,7 @@ int async_scheduler::execute(const value_t &expr, const execute_options &opts) { proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); proc.state = evaluator_.create_state(expr, opts.env); + proc.state.data_step_nodes = data_settings().step_nodes; proc.state.stats.start(); processes_.emplace(pid, std::make_shared(std::move(proc))); return pid; @@ -148,9 +150,11 @@ void async_scheduler::execute_process_step(process &proc) { } reset_guard{current_pid_, current_proc_}; // A process still reading its script (parser.step_tokens) takes signals and watch events once - // the script is read: their handlers run on the evaluation stack, which doesn't exist yet. - const bool can_interrupt = - !proc.in_signal_handler && !proc.in_watch_handler && !proc.state.reading; + // the script is read: their handlers run on the evaluation stack, which doesn't exist yet. One + // in the middle of deep-data work (data.step_nodes) takes them once that is done: the work's + // call frame waits on top of the stack. + const bool can_interrupt = !proc.in_signal_handler && !proc.in_watch_handler && + !proc.state.reading && !proc.state.data_work; if (!proc.pending_signals.empty() && can_interrupt) { handle_signal(proc); @@ -164,6 +168,7 @@ void async_scheduler::execute_process_step(process &proc) { proc.status = process_status::running; proc.last_run_start = std::chrono::steady_clock::now(); + const auto step_start = proc.last_run_start; // Arm the cooperative deadline for this step so an uncapped nested evaluator or a // long-running builtin aborts at the process's remaining time budget too — the same @@ -215,6 +220,8 @@ void async_scheduler::execute_process_step(process &proc) { proc.accumulated_time += std::chrono::duration(now - proc.last_run_start).count(); if (charge_cpu) proc.charge_cpu_time(thread_cpu_seconds() - cpu0); + // Charge the step's time to what it was spent on (eval, read or data). + proc.state.stats.add_step_time(std::chrono::duration(now - step_start).count()); if ((proc.in_signal_handler || proc.in_watch_handler) && proc.state.done) { restore_from_signal(proc); @@ -720,6 +727,7 @@ int async_scheduler::fork(int pid) { child->state.done = parent.state.done; child->state.stack = parent.state.stack; child->state.user_macros = parent.state.user_macros; + child->state.data_step_nodes = parent.state.data_step_nodes; auto child_env = std::make_shared(); if (parent.state.global_env) { child_env->bindings = parent.state.global_env->bindings; @@ -800,6 +808,7 @@ process_info async_scheduler::make_info(const process &proc) const { info.uid = proc.uid; info.gid = proc.gid; info.step_count = proc.step_count(); + fill_budget_usage(info, proc.state.stats); info.elapsed_time = proc.elapsed_time(); info.current_memory = mem_tracker_.current_bytes(proc.pid); info.peak_memory = mem_tracker_.peak_bytes(proc.pid); diff --git a/src/cvc/state/state_exec/builtins.cpp b/src/cvc/state/state_exec/builtins.cpp index ab0bc020e..3491f9aca 100644 --- a/src/cvc/state/state_exec/builtins.cpp +++ b/src/cvc/state/state_exec/builtins.cpp @@ -1,5 +1,7 @@ +#include #include #include +#include #include #include #include // SE-1: (length "...") counts codepoints for a string @@ -275,12 +277,17 @@ value_t builtin_ge(std::span args) { value_t builtin_eq(std::span args) { expect_exact(args, 2, "="); - return value_t{values_equal(args[0], args[1])}; + if (!is_compound(args[0]) || !is_compound(args[1])) + return value_t{values_equal(args[0], args[1])}; + return run_data_job(make_equal_job(args[0], args[1])); } value_t builtin_ne(std::span args) { expect_exact(args, 2, "!="); - return value_t{!values_equal(args[0], args[1])}; + if (!is_compound(args[0]) || !is_compound(args[1])) + return value_t{!values_equal(args[0], args[1])}; + return run_data_job(make_equal_job(args[0], args[1]), + [](value_t equal) { return value_t{!std::get(equal.v)}; }); } // --------------------------------------------------------------------------- @@ -288,6 +295,8 @@ value_t builtin_ne(std::span args) { // --------------------------------------------------------------------------- value_t builtin_str_concat(std::span args) { + if (std::any_of(args.begin(), args.end(), is_compound)) + return run_data_job(make_concat_job({args.begin(), args.end()}, "", "str-concat")); // The same string may be passed many times, so the result can outgrow every argument: check // each part against the allocation limit before appending it. std::string result; @@ -308,6 +317,8 @@ value_t builtin_str_concat(std::span args) { value_t builtin_str(std::span args) { expect_exact(args, 1, "str"); + if (is_compound(args[0])) + return run_data_job(make_print_job(args[0])); return value_t{to_string(args[0])}; } @@ -634,6 +645,18 @@ value_t builtin_type_of(std::span args) { // --------------------------------------------------------------------------- value_t builtin_print(std::span args) { + if (std::any_of(args.begin(), args.end(), is_compound)) { + // The line goes out once all of it is printed. A string prints quoted here, as to_string + // prints it, so it goes in already printed. + std::vector parts; + parts.reserve(args.size()); + for (auto &a : args) + parts.push_back(std::holds_alternative(a.v) ? value_t{to_string(a)} : a); + return run_data_job(make_concat_job(std::move(parts), " ", "print"), [](value_t line) { + std::cout << std::get(line.v) << '\n'; + return nil_value; + }); + } for (size_t i = 0; i < args.size(); ++i) { if (i > 0) std::cout << ' '; diff --git a/src/cvc/state/state_exec/data_job.cpp b/src/cvc/state/state_exec/data_job.cpp new file mode 100644 index 000000000..2a9dc882e --- /dev/null +++ b/src/cvc/state/state_exec/data_job.cpp @@ -0,0 +1,621 @@ +/* + Copyright 2026 The University of Texas at Austin + + data_job — walks over nested values (compare, print, copy, macro expansion) that keep their + work on the heap and run a slice at a time. +*/ + +#include +#include +#include +#include +#include +#include // evaluation_stats +#include +#include // the settings' key prefixes +#include +#include +#include +#include +#include +#include + +namespace cvc::state_exec { + +namespace { + +// A list or dict being walked, held by owning pointer and read by index. +struct cursor { + list_ptr list; + dict_ptr dict; + std::size_t next = 0; // the next element to visit + + std::size_t size() const { return list ? list->size() : dict->size(); } +}; + +// The cursor for `v` if it is a non-empty list or dict. +std::optional open_cursor(const value_t &v) { + if (auto *l = std::get_if(&v.v); l && *l && !(*l)->empty()) + return cursor{*l, nullptr, 0}; + if (auto *d = std::get_if(&v.v); d && *d && !(*d)->empty()) + return cursor{nullptr, *d, 0}; + return std::nullopt; +} + +// -- print ------------------------------------------------------------------- + +// to_string caps total OUTPUT length (a shared/wide structure would otherwise expand +// exponentially, a cyclic one forever); past it, it emits an ellipsis and stops -- enough for +// debug / coercion / error messages, and impossible to weaponise into a hang or OOM. Depth +// needs no cap of its own: the walk keeps its open lists and dicts on a heap stack, not the +// native one, and each level opened costs output, so the length cap bounds that stack too. +constexpr std::size_t kToStringMaxLen = std::size_t{1} << 16; // 64 KiB +constexpr const char *kEllipsis = "\xE2\x80\xA6"; // U+2026 HORIZONTAL ELLIPSIS + +class print_walk final : public data_job { +public: + explicit print_walk(value_t v) : root_(std::move(v)) {} + + bool step(std::size_t budget) override { + if (!started_) { + started_ = true; + print(root_); + } + for (std::size_t visited = 0; !open_.empty(); ++visited) { + if (budget != 0 && visited >= budget) + return false; + cursor &c = open_.back(); + const bool is_list = c.list != nullptr; + const std::size_t i = c.next++; + if (i >= c.size()) { + out_ += is_list ? ')' : '}'; + open_.pop_back(); + continue; + } + ++nodes_; + if (i > 0) + out_ += is_list ? " " : ", "; + if (out_.size() >= kToStringMaxLen) { + out_ += kEllipsis; + out_ += is_list ? ')' : '}'; + open_.pop_back(); + continue; + } + if (is_list) { + print((*c.list)[i]); // may push, invalidating c + } else { + const auto &[key, value] = (*c.dict)[i]; + out_ += '"'; + out_ += key; + out_ += "\": "; + print(value); // may push, invalidating c + } + } + return true; + } + + value_t result() override { return value_t{std::move(out_)}; } + std::string take_string() { return std::move(out_); } + +private: + // Print `val`; a non-empty list or dict is only opened (its bracket printed and a cursor + // pushed for step() to print its elements and close it). + void print(const value_t &val) { + if (out_.size() >= kToStringMaxLen) { + out_ += kEllipsis; + return; + } + std::visit( + [&](auto &&arg) { + using T = std::decay_t; + if constexpr (std::is_same_v) + out_ += "nil"; + else if constexpr (std::is_same_v) + out_ += arg ? "#t" : "#f"; + else if constexpr (std::is_same_v) + out_ += std::to_string(arg); + else if constexpr (std::is_same_v) { + std::ostringstream oss; + oss << arg; + out_ += oss.str(); + } else if constexpr (std::is_same_v) { + out_ += '"'; + out_ += arg; + out_ += '"'; + } else if constexpr (std::is_same_v) { + // b"..." — printable ASCII verbatim, everything else (incl. all bytes >= 0x80) + // hex-escaped as \xNN, so the rendering is always valid ASCII text and never dumps + // raw binary bytes. + static const char kHex[] = "0123456789abcdef"; + out_ += "b\""; + for (unsigned char b : arg.data) { + if (b == '"' || b == '\\') { + out_ += '\\'; + out_ += static_cast(b); + } else if (b >= 0x20 && b < 0x7F) { + out_ += static_cast(b); + } else { + out_ += "\\x"; + out_ += kHex[b >> 4]; + out_ += kHex[b & 0x0F]; + } + } + out_ += '"'; + } else if constexpr (std::is_same_v) + out_ += arg.name; + else if constexpr (std::is_same_v) { + if (!arg || arg->empty()) { + out_ += "()"; + return; + } + out_ += '('; + open_.push_back({arg, nullptr, 0}); + } else if constexpr (std::is_same_v) + out_ += ""; + else if constexpr (std::is_same_v) { + if (!arg || arg->empty()) { + out_ += "{}"; + return; + } + out_ += '{'; + open_.push_back({nullptr, arg, 0}); + } else if constexpr (std::is_same_v) + out_ += ""; + else if constexpr (std::is_same_v) + out_ += arg ? "type_name + ">" : ""; + else if constexpr (std::is_same_v) + out_ += arg && arg->exhausted ? "" : ""; + else + out_ += ""; + }, + val.v); + } + + value_t root_; + bool started_ = false; + std::string out_; + std::vector open_; +}; + +class concat_walk final : public data_job { +public: + concat_walk(std::vector parts, std::string sep, const char *who) + : parts_(std::move(parts)), sep_(std::move(sep)), who_(who) {} + + bool step(std::size_t budget) override { + std::size_t used = 0; + while (next_ < parts_.size()) { + if (!part_) { + if (next_ > 0) + append(sep_); + if (auto *s = std::get_if(&parts_[next_].v)) { + append(*s); + ++next_; + continue; + } + part_ = std::make_unique(parts_[next_]); + } + if (budget != 0 && used >= budget) + return false; + const std::size_t before = part_->nodes(); + const bool done = part_->step(budget == 0 ? 0 : budget - used); + used += std::max(1, part_->nodes() - before); + nodes_ += part_->nodes() - before; + if (!done) + return false; + append(part_->take_string()); + part_.reset(); + ++next_; + } + return true; + } + + value_t result() override { return value_t{std::move(out_)}; } + +private: + void append(const std::string &part) { + eval_alloc_check(out_.size() > SIZE_MAX - part.size() ? SIZE_MAX : out_.size() + part.size(), + who_); + out_ += part; + } + + std::vector parts_; + std::string sep_; + const char *who_; + std::size_t next_ = 0; + std::unique_ptr part_; // the part being printed + std::string out_; +}; + +// -- equal ------------------------------------------------------------------- + +struct ptr_pair_hash { + std::size_t operator()(const std::pair &p) const noexcept { + return std::hash()(p.first) * 1000003u ^ std::hash()(p.second); + } +}; +using ptr_pair_set = std::unordered_set, ptr_pair_hash>; + +// `seen_` memoizes the container pairs already under comparison: re-encountering one (a shared +// node reached by two paths, or a cycle) counts as equal co-inductively. That bounds the work to +// the DISTINCT node pairs (the physical size), never the exponential logical unfolding of a +// shared DAG, and ends on cyclic structures; for finite acyclic values the result is that of a +// naive compare. A deadline poll per node bounds a huge NON-shared structure (memoization alone +// does not: distinct nodes are never re-encountered). +class equal_walk final : public data_job { +public: + equal_walk(value_t a, value_t b) : a_(std::move(a)), b_(std::move(b)) {} + + bool step(std::size_t budget) override { + if (done_) + return true; + if (!started_) { + started_ = true; + if (!visit(a_, b_)) + return finish(false); + } + for (std::size_t visited = 0; !open_.empty(); ++visited) { + if (budget != 0 && visited >= budget) + return false; + auto &[ca, cb] = open_.back(); + const std::size_t n = ca.size(); + if (n != cb.size()) // changed between two slices + return finish(false); + const std::size_t i = ca.next++; + if (i >= n) { + open_.pop_back(); + continue; + } + ++nodes_; + if (eval_deadline_expired()) + throw std::runtime_error("values_equal: structure too large to compare"); + bool equal; + // Element references stay valid through visit(), which only pushes onto open_. + if (ca.list) { + const value_t &x = (*ca.list)[i]; + const value_t &y = (*cb.list)[i]; + equal = visit(x, y); // may push, invalidating ca and cb + } else { + const auto &[kx, x] = (*ca.dict)[i]; + const auto &[ky, y] = (*cb.dict)[i]; + equal = kx == ky && visit(x, y); // may push, invalidating ca and cb + } + if (!equal) + return finish(false); + } + return finish(true); + } + + value_t result() override { return value_t{equal_}; } + bool equal() const { return equal_; } + +private: + bool finish(bool equal) { + equal_ = equal; + done_ = true; + open_.clear(); + return true; + } + + // Whether x and y are equal as far as one level shows; a list or dict pair still to compare + // element by element gets a cursor pair. + bool visit(const value_t &x, const value_t &y) { + if (x.v.index() != y.v.index()) + return false; + return std::visit( + [&](auto &&arg_a) -> bool { + using T = std::decay_t; + auto &arg_b = std::get(y.v); + if constexpr (std::is_same_v) + return true; + else if constexpr (std::is_same_v || std::is_same_v) { + if (arg_a == arg_b) + return true; + if (!arg_a || !arg_b) + return false; + if (arg_a->size() != arg_b->size()) + return false; + if (!seen_) + seen_.emplace(); // not before the first pair: comparing scalars allocates nothing + if (!seen_ + ->insert({static_cast(arg_a.get()), + static_cast(arg_b.get())}) + .second) + return true; // this pair is already being compared (shared/cyclic) -> equal + if constexpr (std::is_same_v) + open_.push_back({cursor{arg_a, nullptr, 0}, cursor{arg_b, nullptr, 0}}); + else + open_.push_back({cursor{nullptr, arg_a, 0}, cursor{nullptr, arg_b, 0}}); + return true; + } else if constexpr (std::is_same_v) + return false; // functions are never equal + else + // bool, numbers, strings, symbols and bytes by value; closures, data objects and + // generators by identity + return arg_a == arg_b; + }, + x.v); + } + + value_t a_, b_; + bool started_ = false; + bool done_ = false; + bool equal_ = false; + std::optional seen_; + std::vector> open_; +}; + +// -- copy -------------------------------------------------------------------- + +// Memoized on the source node, so a shared-pointer DAG is copied in PHYSICAL time (the copy has +// the same sharing, disjoint from the original) and a cycle is copied as a cycle. A deadline +// poll per node aborts a huge payload at the time budget. +class copy_walk final : public data_job { +public: + explicit copy_walk(value_t v) : root_(std::move(v)) {} + + bool step(std::size_t budget) override { + if (!started_) { + started_ = true; + result_ = copy(root_); + } + for (std::size_t visited = 0; !open_.empty(); ++visited) { + if (budget != 0 && visited >= budget) + return false; + auto &[src, dst] = open_.back(); + const std::size_t i = src.next++; + if (i >= src.size()) { + open_.pop_back(); + continue; + } + ++nodes_; + if (eval_deadline_expired()) + throw std::runtime_error("state-data-get: evaluation exceeded time budget"); + if (src.list) { + list_ptr out = std::get(dst.v); + const value_t &e = (*src.list)[i]; + out->push_back(copy(e)); // may push, invalidating src and dst + } else { + dict_ptr out = std::get(dst.v); + const auto &[k, e] = (*src.dict)[i]; + out->emplace_back(k, copy(e)); // may push, invalidating src and dst + } + } + return true; + } + + value_t result() override { return std::move(result_); } + +private: + // The copy of `v`: a list or dict is a new empty one, filled by step() through its cursor. + value_t copy(const value_t &v) { + std::optional c = open_cursor(v); + if (!c) { + // empty or null lists and dicts still get their own (empty) copy + if (auto *l = std::get_if(&v.v); l && *l) + return make_list(); + if (auto *d = std::get_if(&v.v); d && *d) + return make_dict(); + return v; // scalar / symbol (opaque callables are refused at state-data-set) + } + const void *key = c->list ? static_cast(c->list.get()) + : static_cast(c->dict.get()); + if (auto found = memo_.find(key); found != memo_.end()) + return found->second; // shared node / cycle already copied -> reuse (bounds the walk) + value_t out; + if (c->list) { + auto l = std::make_shared>(); + l->reserve(c->list->size()); + out = value_t{l}; + } else { + auto d = std::make_shared(); + d->reserve(c->dict->size()); + out = value_t{d}; + } + memo_.emplace(key, out); // registered before its elements are copied + open_.emplace_back(std::move(*c), out); + return out; + } + + value_t root_; + bool started_ = false; + value_t result_; + std::unordered_map memo_; + std::vector> open_; +}; + +// -- substitute -------------------------------------------------------------- + +class substitute_walk final : public data_job { +public: + substitute_walk(value_t tmpl, std::unordered_map subst) + : root_(std::move(tmpl)), subst_(std::move(subst)) {} + + bool step(std::size_t budget) override { + if (!started_) { + started_ = true; + result_ = expand(root_); + } + for (std::size_t visited = 0; !open_.empty(); ++visited) { + if (budget != 0 && visited >= budget) + return false; + auto &[src, dst] = open_.back(); + const std::size_t i = src.next++; + if (i >= src.list->size()) { + open_.pop_back(); + continue; + } + ++nodes_; + list_ptr out = dst; + out->push_back(expand((*src.list)[i])); // may push, invalidating src and dst + } + return true; + } + + value_t result() override { return std::move(result_); } + +private: + value_t expand(const value_t &v) { + if (auto *sym = std::get_if(&v.v)) { + auto it = subst_.find(sym->name); + return it != subst_.end() ? it->second : v; + } + if (auto *l = std::get_if(&v.v); l && *l) { + auto out = std::make_shared>(); + out->reserve((*l)->size()); + open_.emplace_back(cursor{*l, nullptr, 0}, out); + return value_t{out}; + } + return v; + } + + value_t root_; + std::unordered_map subst_; + bool started_ = false; + value_t result_; + std::vector> open_; +}; + +// The data side of the evaluator step running on this thread, if any. +thread_local detail::data_step_scope *t_scope = nullptr; + +// Run `walk` to completion and count its nodes into the current step. +template Walk &finish_walk(Walk &walk) { + walk.step(0); + count_data_nodes(walk.nodes()); + return walk; +} + +std::optional read_size_setting(cvc::state &root, const std::string &path) { + cvc::state *node = root.findDescendant(path); + if (!node) + return std::nullopt; + const std::string v = node->value(); + std::size_t n = 0; + const auto [end, ec] = std::from_chars(v.data(), v.data() + v.size(), n); + if (ec != std::errc{} || end != v.data() + v.size() || v.empty()) + return std::nullopt; + return n; +} + +} // namespace + +// -- the walks as free functions ---------------------------------------------- + +std::string to_string(const value_t &val) { + print_walk walk(val); + return finish_walk(walk).take_string(); +} + +bool values_equal(const value_t &a, const value_t &b) { + equal_walk walk(a, b); + return finish_walk(walk).equal(); +} + +// -- jobs ---------------------------------------------------------------------- + +bool is_compound(const value_t &v) { + if (auto *l = std::get_if(&v.v)) + return *l && !(*l)->empty(); + if (auto *d = std::get_if(&v.v)) + return *d && !(*d)->empty(); + return false; +} + +std::unique_ptr make_concat_job(std::vector parts, std::string sep, + const char *who) { + return std::make_unique(std::move(parts), std::move(sep), who); +} + +std::unique_ptr make_equal_job(value_t a, value_t b) { + return std::make_unique(std::move(a), std::move(b)); +} + +std::unique_ptr make_print_job(value_t v) { + return std::make_unique(std::move(v)); +} + +std::unique_ptr make_copy_job(value_t v) { + return std::make_unique(std::move(v)); +} + +std::unique_ptr make_substitute_job(value_t tmpl, + std::unordered_map subst) { + return std::make_unique(std::move(tmpl), std::move(subst)); +} + +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(); +} + +// -- settings ------------------------------------------------------------------ + +data_options data_options_from_state(cvc::state &root, const std::string &scheduler_id, + const data_options &base) { + data_options opts = base; + const auto resolve = [&](const char *key, std::size_t &out) { + const std::string suffix = std::string(".data.") + key; + if (auto v = read_size_setting(root, std::string(scheduler::defaults_prefix) + suffix)) + out = *v; + if (!scheduler_id.empty()) + if (auto v = read_size_setting(root, std::string(scheduler::sched_prefix) + "." + + scheduler_id + suffix)) + out = *v; + }; + resolve("step_nodes", opts.step_nodes); // 0 = no slicing + return opts; +} + +// -- running a native call's job ------------------------------------------------- + +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; + } + value_t r = job->result(); + return then ? then(std::move(r)) : r; +} + +void count_data_nodes(std::size_t nodes) { + if (nodes == 0) + return; + if (detail::data_step_scope *scope = t_scope) { + scope->nodes_ += nodes; + scope->stats_.add_data_nodes(nodes); + } +} + +// -- data_step_scope ------------------------------------------------------------- + +detail::data_step_scope::data_step_scope(std::size_t step_nodes, evaluation_stats &stats, + release_backlog *backlog) + : step_nodes_(step_nodes), stats_(stats), backlog_(step_nodes > 0 ? backlog : nullptr), + outer_(t_scope), routing_(backlog_) { + t_scope = this; +} + +detail::data_step_scope::~data_step_scope() { t_scope = outer_; } + +std::size_t detail::data_step_scope::remaining() const { + return step_nodes_ > nodes_ ? step_nodes_ - nodes_ : 1; +} + +void detail::data_step_scope::drain_backlog() { + if (!backlog_ || backlog_->empty() || nodes_ >= step_nodes_) + return; + count_data_nodes(backlog_->drain(step_nodes_ - nodes_)); +} + +detail::data_step_scope *detail::data_step_scope::current() { return t_scope; } + +} // namespace cvc::state_exec diff --git a/src/cvc/state/state_exec/intrinsics.cpp b/src/cvc/state/state_exec/intrinsics.cpp index 43ca8ce6c..de35c0652 100644 --- a/src/cvc/state/state_exec/intrinsics.cpp +++ b/src/cvc/state/state_exec/intrinsics.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -88,57 +89,6 @@ std::string coerce_state_string(const value_t &v) { return to_string(v); // list/dict/symbol/closure — readable fallback } -// Clone a value's compound structure (list/dict) so the caller gets a PRIVATE copy that does -// not alias node-resident storage. Bounded and cycle-safe, mirroring values_equal / to_string -// (it is the third structural walker and must not be the weak link): it MEMOIZES on the source -// node pointer, so a shared-pointer DAG is copied in PHYSICAL time (the copy preserves the same -// sharing, disjoint from the stored original) and a cycle terminates; it keeps the elements -// still to copy on a heap stack, so any depth copies without native recursion; and it polls -// the evaluation deadline so a huge payload aborts at the time budget instead of running to -// completion in one native step. -value_t deep_copy(const value_t &v) { - std::unordered_map memo; - value_t result; - // (source, destination) pairs: each destination is a slot of a copy already allocated at its - // final size, so it stays put while the copy fills in. - std::vector> pending{{&v, &result}}; - while (!pending.empty()) { - if (eval_deadline_expired()) - throw std::runtime_error("state-data-get: evaluation exceeded time budget"); - auto [src, dst] = pending.back(); - pending.pop_back(); - if (auto *l = std::get_if(&src->v); l && *l) { - auto found = memo.find(static_cast(l->get())); - if (found != memo.end()) { - *dst = found->second; // shared node / cycle already copied -> reuse (bounds the walk) - continue; - } - auto out = std::make_shared>((*l)->size()); - *dst = value_t{out}; - memo.emplace(static_cast(l->get()), *dst); // register BEFORE descending - for (std::size_t i = (*l)->size(); i-- > 0;) - pending.emplace_back(&(**l)[i], &(*out)[i]); - } else if (auto *dp = std::get_if(&src->v); dp && *dp) { - auto found = memo.find(static_cast(dp->get())); - if (found != memo.end()) { - *dst = found->second; - continue; - } - auto out = std::make_shared>>(); - out->reserve((*dp)->size()); - for (const auto &kv : **dp) - out->emplace_back(kv.first, value_t{}); - *dst = value_t{out}; - memo.emplace(static_cast(dp->get()), *dst); - for (std::size_t i = (*dp)->size(); i-- > 0;) - pending.emplace_back(&(**dp)[i].second, &(*out)[i].second); - } else { - *dst = *src; // scalar / symbol / null (opaque callables are refused at state-data-set) - } - } - return result; -} - void require_root(const intrinsics_context *ctx, const char *name) { if (!ctx->root) throw std::runtime_error(std::string(name) + ": no state root bound"); @@ -249,7 +199,7 @@ value_t intrinsic_state_data_get(intrinsics_context *ctx, std::span(&d)) - return deep_copy(*v); + return run_data_job(make_copy_job(*v)); // A RAW std::string blob on the data channel reads back as the DSL `bytes` type (opaque octets), // NOT an opaque data_object. That raw representation is what state-data-set stores for a `bytes` // value, what the `?data` URI channel writes (uri_state.cpp), and what the §13.9 HTTP cache parks @@ -461,6 +411,18 @@ value_t intrinsic_ps(intrinsics_context *ctx, std::span args) { return make_list(std::move(result)); } +// The budget-by-kind fields of `info`, as dict entries. +static void append_budget_usage(std::vector> &entries, + const process_info &info) { + entries.emplace_back("read_steps", value_t(static_cast(info.read_steps))); + entries.emplace_back("eval_steps", value_t(static_cast(info.eval_steps))); + entries.emplace_back("data_steps", value_t(static_cast(info.data_steps))); + entries.emplace_back("data_nodes", value_t(static_cast(info.data_nodes))); + entries.emplace_back("read_time", value_t(info.read_time)); + entries.emplace_back("eval_time", value_t(info.eval_time)); + entries.emplace_back("data_time", value_t(info.data_time)); +} + value_t intrinsic_inspect(intrinsics_context *ctx, std::span args) { expect_exact(args, 1, "inspect"); require_sched(ctx, "inspect"); @@ -483,6 +445,7 @@ value_t intrinsic_inspect(intrinsics_context *ctx, std::span args entries.emplace_back("uid", value_t(info->uid)); entries.emplace_back("gid", value_t(info->gid)); entries.emplace_back("steps", value_t(static_cast(info->step_count))); + append_budget_usage(entries, *info); entries.emplace_back("elapsed_time", value_t(info->elapsed_time)); entries.emplace_back("memory", value_t(static_cast(info->current_memory))); entries.emplace_back("peak_memory", value_t(static_cast(info->peak_memory))); @@ -556,6 +519,25 @@ value_t intrinsic_message_bytes_limit(intrinsics_context *ctx, std::span(ctx->proc->max_message_bytes)); } +// What the current process's budget went on: steps, seconds and deep-data nodes by kind (read, +// eval, data), plus the total steps -- the same fields inspect reports for any pid. +value_t intrinsic_budget_usage(intrinsics_context *ctx, std::span args) { + expect_exact(args, 0, "budget-usage"); + // The running process, as self-pid finds it: the scheduler's current one, else the context's. + process_info info{}; + if (ctx->sched && ctx->sched->current_pid() >= 0) { + if (auto running = ctx->sched->get_process_info(ctx->sched->current_pid())) + info = *running; + } else if (ctx->proc) { + info.step_count = ctx->proc->step_count(); + fill_budget_usage(info, ctx->proc->state.stats); + } + std::vector> entries; + entries.emplace_back("steps", value_t(static_cast(info.step_count))); + append_budget_usage(entries, info); + return make_dict(std::move(entries)); +} + value_t intrinsic_step_count(intrinsics_context *ctx, std::span args) { expect_exact(args, 0, "step-count"); if (!ctx->proc) @@ -872,6 +854,7 @@ void register_intrinsics(environment_ptr env, intrinsics_context *ctx) { reg("message-bytes", intrinsic_message_bytes); reg("message-bytes-limit", intrinsic_message_bytes_limit); reg("step-count", intrinsic_step_count); + reg("budget-usage", intrinsic_budget_usage); // System identity reg("cluster-id", intrinsic_cluster_id); diff --git a/src/cvc/state/state_exec/scheduler.cpp b/src/cvc/state/state_exec/scheduler.cpp index b3ae0f1aa..019f3be5d 100644 --- a/src/cvc/state/state_exec/scheduler.cpp +++ b/src/cvc/state/state_exec/scheduler.cpp @@ -55,8 +55,9 @@ void scheduler::load_settings() { max_pending_messages = effective; - // --- parser.* (same resolution order; see parse_options_from_state) --- + // --- parser.* and data.* (same resolution order; see parse_options_from_state) --- parse_options_ = parse_options_from_state(*watch_root_, id_); + data_options_ = data_options_from_state(*watch_root_, id_); // --- limits.max_alloc_bytes (same resolution order; see max_alloc_bytes_from_state) --- max_alloc_bytes_ = max_alloc_bytes_from_state(*watch_root_, id_); @@ -66,6 +67,7 @@ void scheduler::load_settings() { (*watch_root_)(base + ".max_pending_messages").value(std::to_string(max_pending_messages)); (*watch_root_)(base + ".parser.max_depth").value(std::to_string(parse_options_.max_depth)); (*watch_root_)(base + ".parser.step_tokens").value(std::to_string(parse_options_.step_tokens)); + (*watch_root_)(base + ".data.step_nodes").value(std::to_string(data_options_.step_nodes)); (*watch_root_)(base + ".limits.max_alloc_bytes").value(std::to_string(max_alloc_bytes_)); (*watch_root_)(base + ".policy") .value(policy_ == scheduling_policy::round_robin ? "round_robin" @@ -112,6 +114,7 @@ int scheduler::execute(const std::string &script, const execute_options &opts) { proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); proc.state = evaluator_.create_state(script, opts.env, parse_options_); + proc.state.data_step_nodes = data_options_.step_nodes; proc.state.stats.start(); processes_.emplace(pid, std::make_shared(std::move(proc))); return pid; @@ -138,6 +141,7 @@ int scheduler::execute(const value_t &expr, const execute_options &opts) { proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); proc.state = evaluator_.create_state(expr, opts.env); + proc.state.data_step_nodes = data_options_.step_nodes; proc.state.stats.start(); processes_.emplace(pid, std::make_shared(std::move(proc))); return pid; @@ -215,9 +219,11 @@ void scheduler::execute_process_step(process &proc) { } reset_guard{current_pid_, current_proc_}; // A process still reading its script (parser.step_tokens) takes signals and watch events once - // the script is read: their handlers run on the evaluation stack, which doesn't exist yet. - const bool can_interrupt = - !proc.in_signal_handler && !proc.in_watch_handler && !proc.state.reading; + // the script is read: their handlers run on the evaluation stack, which doesn't exist yet. One + // in the middle of deep-data work (data.step_nodes) takes them once that is done: the work's + // call frame waits on top of the stack. + const bool can_interrupt = !proc.in_signal_handler && !proc.in_watch_handler && + !proc.state.reading && !proc.state.data_work; // 1. Handle pending signals first if (!proc.pending_signals.empty() && can_interrupt) { @@ -254,9 +260,11 @@ void scheduler::execute_process_step(process &proc) { const double cpu0 = charge_cpu ? thread_cpu_seconds() : 0.0; bool done = evaluator_.step(proc.state); - // 4. Accumulate running time + // 4. Accumulate running time, and charge it to what the step was spent on auto now = std::chrono::steady_clock::now(); - proc.accumulated_time += std::chrono::duration(now - proc.last_run_start).count(); + const double step_seconds = std::chrono::duration(now - proc.last_run_start).count(); + proc.accumulated_time += step_seconds; + proc.state.stats.add_step_time(step_seconds); if (charge_cpu) proc.charge_cpu_time(thread_cpu_seconds() - cpu0); @@ -723,6 +731,7 @@ int scheduler::fork(int pid) { child->state.done = parent.state.done; child->state.stack = parent.state.stack; // Copy stack child->state.user_macros = parent.state.user_macros; + child->state.data_step_nodes = parent.state.data_step_nodes; // Create fresh env as a copy of parent's bindings auto child_env = std::make_shared(); if (parent.state.global_env) { @@ -806,6 +815,7 @@ process_info scheduler::make_info(const process &proc) const { info.uid = proc.uid; info.gid = proc.gid; info.step_count = proc.step_count(); + fill_budget_usage(info, proc.state.stats); info.elapsed_time = proc.elapsed_time(); info.current_memory = mem_tracker_.current_bytes(proc.pid); info.peak_memory = mem_tracker_.peak_bytes(proc.pid); diff --git a/src/cvc/state/state_exec/stackless_evaluator.cpp b/src/cvc/state/state_exec/stackless_evaluator.cpp index a1483a6a3..7858da2b8 100644 --- a/src/cvc/state/state_exec/stackless_evaluator.cpp +++ b/src/cvc/state/state_exec/stackless_evaluator.cpp @@ -74,25 +74,110 @@ evaluator_state stackless_evaluator::create_program_state(std::vector f // parse_error propagates like any error from step(): the sync scheduler rethrows it to its // caller, and the async scheduler kills the process with it. bool stackless_evaluator::step_read(evaluator_state &state) { - state.stats.increment_step(); + state.stats.increment_step(step_kind::read); script_reader &reader = *state.reading; if (!reader.step(reader.options().step_tokens)) return false; evaluator_state program = create_program_state(std::move(reader.forms()), state.global_env); program.stats = std::move(state.stats); + program.data_step_nodes = state.data_step_nodes; + program.backlog = std::move(state.backlog); state = std::move(program); // also releases the reader return state.done; } +namespace { +// RAII: while alive, the native call a step applies may leave deep-data work behind +// (run_data_job), in the step's data scope. +struct native_call_scope { + explicit native_call_scope(detail::data_step_scope *data) : data(data) { + if (data) + data->allow_defer(true); + } + ~native_call_scope() { + if (data) + data->allow_defer(false); + } + native_call_scope(const native_call_scope &) = delete; + native_call_scope &operator=(const native_call_scope &) = delete; + detail::data_step_scope *data; +}; +} // 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. +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 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(); + if (work->what == pending_data_work::kind::macro_expansion) { + begin_expansion(state, std::move(result)); + return false; + } + if (work->then) { + 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}); + return false; + } + } + pop_frame(state, std::move(result)); + return state.done; +} + +void stackless_evaluator::begin_expansion(evaluator_state &state, value_t expansion) { + eval_frame &frame = state.stack.back(); + frame.expr = std::move(expansion); + frame.phase = eval_phase::init; + frame.index = 0; + frame.results.clear(); + frame.extra_vals.clear(); + frame.extra_strs.clear(); + frame.extra_lists.clear(); + frame.extra_env = nullptr; +} + bool stackless_evaluator::step(evaluator_state &state) { check_interrupted(); - if (state.reading) - return step_read(state); - if (state.done || state.stack.empty()) { + if (state.data_step_nodes > 0 && !state.backlog) + state.backlog = std::make_shared(); + // Held here so the backlog outlives this step's scope even if the step replaces the state. + const std::shared_ptr backlog = state.backlog; + detail::data_step_scope data(state.data_step_nodes, state.stats, backlog.get()); + bool done; + if (state.reading) { + done = step_read(state); + } else if (state.data_work) { + done = step_data(state); + } else if (backlog && !backlog->empty()) { + // A step that only frees what earlier steps dropped. + state.stats.increment_step(step_kind::data); + data.drain_backlog(); + return false; + } else if (state.done || state.stack.empty()) { state.done = true; return true; + } else { + state.stats.increment_step(); + step_frame(state); + done = state.done; } - state.stats.increment_step(); + data.drain_backlog(); + // Done only once what it dropped is freed too: the steps that free the rest come first. + return done && !(backlog && !backlog->empty()); +} + +void stackless_evaluator::step_frame(evaluator_state &state) { auto &frame = state.stack.back(); switch (frame.phase) { @@ -150,7 +235,6 @@ bool stackless_evaluator::step(evaluator_state &state) { step_break_unwind(state, frame); break; } - return state.done; } value_t stackless_evaluator::run(evaluator_state &state, std::optional max_steps, @@ -524,15 +608,18 @@ void stackless_evaluator::step_init(evaluator_state &state, eval_frame &frame) { std::unordered_map subst; for (size_t i = 0; i < macro_params.size(); ++i) subst[macro_params[i].name] = (i < args.size()) ? args[i] : nil_value; - // Re-init current frame with expanded form for evaluation - frame.expr = substitute(body, subst); - frame.phase = eval_phase::init; - frame.index = 0; - frame.results.clear(); - frame.extra_vals.clear(); - frame.extra_strs.clear(); - frame.extra_lists.clear(); - frame.extra_env = nullptr; + // Re-init current frame with expanded form for evaluation. The expansion is deep-data work: + // a step that slices does one slice of it here and leaves the rest to step_data(). + auto job = make_substitute_job(body, std::move(subst)); + detail::data_step_scope *data = detail::data_step_scope::current(); + 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}); + return; + } + begin_expansion(state, job->result()); return; } @@ -581,7 +668,20 @@ void stackless_evaluator::step_apply(evaluator_state &state, eval_frame &frame) throw std::runtime_error("apply: undefined function: " + head_sym->name); if (auto *fn = std::get_if(&func_val->v)) { - pop_frame(state, (*fn)(evaled)); + detail::data_step_scope *data = detail::data_step_scope::current(); + value_t result; + { + native_call_scope call(data); + result = (*fn)(evaled); + } + 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}); + return; + } + pop_frame(state, std::move(result)); return; } if (auto *cp = std::get_if(&func_val->v)) { @@ -1106,33 +1206,4 @@ void stackless_evaluator::handle_defclass(evaluator_state &state, eval_frame &fr pop_frame(state, *env->lookup(cls_name)); } -// --------------------------------------------------------------------------- -// macro substitution -// --------------------------------------------------------------------------- - -value_t stackless_evaluator::substitute(const value_t &tmpl, - const std::unordered_map &subst) { - // Like the evaluator, the copy keeps its work on the heap, not the native stack, so a - // template of any depth expands. (source, destination) pairs: each destination is a slot of - // a list copy allocated at its final size, so it stays put while the copy fills in. - value_t result; - std::vector> pending{{&tmpl, &result}}; - while (!pending.empty()) { - auto [src, dst] = pending.back(); - pending.pop_back(); - if (auto *sym = std::get_if(&src->v)) { - auto it = subst.find(sym->name); - *dst = it != subst.end() ? it->second : *src; - } else if (auto *lp = std::get_if(&src->v); lp && *lp) { - auto copy = std::make_shared>((*lp)->size()); - *dst = value_t{copy}; - for (std::size_t i = (*lp)->size(); i-- > 0;) - pending.emplace_back(&(**lp)[i], &(*copy)[i]); - } else { - *dst = *src; - } - } - return result; -} - } // namespace cvc::state_exec diff --git a/src/cvc/state/state_exec/state_value_codec.cpp b/src/cvc/state/state_exec/state_value_codec.cpp index 1106f07b4..cd9c0b893 100644 --- a/src/cvc/state/state_exec/state_value_codec.cpp +++ b/src/cvc/state/state_exec/state_value_codec.cpp @@ -708,6 +708,11 @@ void encode_evaluator_state(cvc::state &node, const evaluator_state &es) { r("step_tokens").value(std::to_string(es.reading->options().step_tokens)); } + // Deep-data slicing (data.step_nodes). Work in flight is not kept: after a restore the waiting + // native call or macro expansion runs again. + if (es.data_step_nodes > 0) + node("data_step_nodes").value(std::to_string(es.data_step_nodes)); + // Stats snapshot node("step_count").value(std::to_string(es.stats.get_step_count())); } @@ -756,6 +761,14 @@ evaluator_state decode_evaluator_state(cvc::state &node) { } } + if (has_child(node, "data_step_nodes")) { + const std::string v = node("data_step_nodes").value(); + std::size_t n = 0; + const auto [end, ec] = std::from_chars(v.data(), v.data() + v.size(), n); + if (ec == std::errc{} && end == v.data() + v.size()) + es.data_step_nodes = n; + } + // A script still being read restarts its reading from the beginning. if (has_child(node, "reading")) { auto &r = node("reading"); diff --git a/src/cvc/state/state_exec/types.cpp b/src/cvc/state/state_exec/types.cpp index 709a28304..1bb94300d 100644 --- a/src/cvc/state/state_exec/types.cpp +++ b/src/cvc/state/state_exec/types.cpp @@ -1,11 +1,8 @@ #include -#include #include #include -#include #include #include -#include #include namespace cvc::state_exec { @@ -193,13 +190,24 @@ class release_queue { // inline. A raw pointer to the running release's local queue, so no thread-exit destructor is // involved and nothing outlives the release. thread_local release_queue *t_release_queue = nullptr; + +// The backlog a backlog_scope routes this thread's last owners to, if any. +thread_local detail::release_backlog *t_backlog = nullptr; } // namespace -void detail::release_nested(std::shared_ptr owner) noexcept { +void detail::release_nested(std::shared_ptr owner, std::size_t nodes) noexcept { if (release_queue *queue = t_release_queue) { queue->push(owner); // out of memory: `owner` is still here, and freed here, recursively return; } + if (release_backlog *backlog = t_backlog) { + try { + backlog->owners_.emplace_back(std::move(owner), nodes); + return; + } catch (...) { + // Out of memory: emplace_back left `owner` as it was; free it now, below. + } + } release_queue queue; t_release_queue = &queue; owner.reset(); @@ -208,6 +216,33 @@ void detail::release_nested(std::shared_ptr owner) noexcept { t_release_queue = nullptr; } +detail::release_backlog::~release_backlog() { drain(0); } + +std::size_t detail::release_backlog::drain(std::size_t budget) { + release_backlog *outer = t_backlog; + t_backlog = this; // what freeing an owner drops joins this backlog + std::size_t freed = 0; + while (!owners_.empty() && (budget == 0 || freed < budget)) { + auto [owner, nodes] = std::move(owners_.back()); + owners_.pop_back(); + owner.reset(); + freed += nodes; + } + t_backlog = outer; + return freed; +} + +detail::backlog_scope::backlog_scope(release_backlog *backlog) + : outer_(t_backlog), armed_(backlog != nullptr) { + if (armed_) + t_backlog = backlog; +} + +detail::backlog_scope::~backlog_scope() { + if (armed_) + t_backlog = outer_; +} + // -- per-call allocation limit ---------------------------------------------- namespace { @@ -326,231 +361,6 @@ environment_ptr environment::extend(environment_ptr parent) { return child; } -// -- to_string --------------------------------------------------------------- - -namespace { -// to_string caps total OUTPUT length (a shared/wide structure would otherwise expand -// exponentially, a cyclic one forever); past it, it emits an ellipsis and stops -- enough for -// debug / coercion / error messages, and impossible to weaponise into a hang or OOM. Depth -// needs no cap of its own: the walk keeps its open lists and dicts on a heap stack, not the -// native one, and each level opened costs output, so the length cap bounds that stack too. -constexpr std::size_t kToStringMaxLen = std::size_t{1} << 16; // 64 KiB -constexpr const char *kEllipsis = "\xE2\x80\xA6"; // U+2026 HORIZONTAL ELLIPSIS - -// A list or dict to_string has opened and not yet closed. -struct print_frame { - const std::vector *list = nullptr; - const ordered_map *dict = nullptr; - std::size_t next = 0; // the next element to print -}; - -// Print `val`; a non-empty list or dict is only opened (its bracket printed and a frame pushed -// for to_string's loop to print its elements and close it). -void print_value(const value_t &val, std::string &out, std::vector &open) { - if (out.size() >= kToStringMaxLen) { - out += kEllipsis; - return; - } - std::visit( - [&](auto &&arg) { - using T = std::decay_t; - if constexpr (std::is_same_v) - out += "nil"; - else if constexpr (std::is_same_v) - out += arg ? "#t" : "#f"; - else if constexpr (std::is_same_v) - out += std::to_string(arg); - else if constexpr (std::is_same_v) { - std::ostringstream oss; - oss << arg; - out += oss.str(); - } else if constexpr (std::is_same_v) { - out += '"'; - out += arg; - out += '"'; - } else if constexpr (std::is_same_v) { - // b"..." — printable ASCII verbatim, everything else (incl. all bytes >= 0x80) - // hex-escaped as \xNN, so the rendering is always valid ASCII text and never dumps raw - // binary bytes. - static const char kHex[] = "0123456789abcdef"; - out += "b\""; - for (unsigned char b : arg.data) { - if (b == '"' || b == '\\') { - out += '\\'; - out += static_cast(b); - } else if (b >= 0x20 && b < 0x7F) { - out += static_cast(b); - } else { - out += "\\x"; - out += kHex[b >> 4]; - out += kHex[b & 0x0F]; - } - } - out += '"'; - } else if constexpr (std::is_same_v) - out += arg.name; - else if constexpr (std::is_same_v) { - if (!arg || arg->empty()) { - out += "()"; - return; - } - out += '('; - open.push_back({arg.get(), nullptr, 0}); - } else if constexpr (std::is_same_v) - out += ""; - else if constexpr (std::is_same_v) { - if (!arg || arg->empty()) { - out += "{}"; - return; - } - out += '{'; - open.push_back({nullptr, arg.get(), 0}); - } else if constexpr (std::is_same_v) - out += ""; - else if constexpr (std::is_same_v) - out += arg ? "type_name + ">" : ""; - else if constexpr (std::is_same_v) - out += arg && arg->exhausted ? "" : ""; - else - out += ""; - }, - val.v); -} -} // namespace - -std::string to_string(const value_t &val) { - std::string out; - std::vector open; - print_value(val, out, open); - while (!open.empty()) { - print_frame &f = open.back(); - const std::size_t i = f.next++; - const bool is_list = f.list != nullptr; - if (i == (is_list ? f.list->size() : f.dict->size())) { - out += is_list ? ')' : '}'; - open.pop_back(); - continue; - } - if (i > 0) - out += is_list ? " " : ", "; - if (out.size() >= kToStringMaxLen) { - out += kEllipsis; - out += is_list ? ')' : '}'; - open.pop_back(); - continue; - } - if (is_list) { - print_value((*f.list)[i], out, open); // may push, invalidating f - } else { - const auto &[key, value] = (*f.dict)[i]; - out += '"'; - out += key; - out += "\": "; - print_value(value, out, open); // may push, invalidating f - } - } - return out; -} - -// -- values_equal ------------------------------------------------------------ - -namespace { -struct ptr_pair_hash { - std::size_t operator()(const std::pair &p) const noexcept { - return std::hash()(p.first) * 1000003u ^ std::hash()(p.second); - } -}; -using ptr_pair_set = std::unordered_set, ptr_pair_hash>; -using value_pairs = std::vector>; - -// Whether a and b are equal as far as one level shows; the element pairs a list or dict pair -// still needs compared are pushed onto `pending`. `seen` memoizes the container pairs already -// under comparison: re-encountering one (a shared node reached by two paths, or a cycle) -// counts as equal co-inductively. -bool shallow_equal(const value_t &a, const value_t &b, ptr_pair_set &seen, value_pairs &pending) { - if (a.v.index() != b.v.index()) - return false; - return std::visit( - [&](auto &&arg_a) -> bool { - using T = std::decay_t; - auto &arg_b = std::get(b.v); - if constexpr (std::is_same_v) - return true; - else if constexpr (std::is_same_v) - return arg_a == arg_b; - else if constexpr (std::is_same_v) - return arg_a == arg_b; - else if constexpr (std::is_same_v) - return arg_a == arg_b; - else if constexpr (std::is_same_v) - return arg_a == arg_b; - else if constexpr (std::is_same_v) - return arg_a == arg_b; - else if constexpr (std::is_same_v) { - if (arg_a == arg_b) - return true; - if (!arg_a || !arg_b) - return false; - if (arg_a->size() != arg_b->size()) - return false; - if (!seen.insert({static_cast(arg_a.get()), - static_cast(arg_b.get())}) - .second) - return true; // this pair is already being compared (shared/cyclic) -> equal - for (std::size_t i = arg_a->size(); i-- > 0;) - pending.emplace_back(&(*arg_a)[i], &(*arg_b)[i]); - return true; - } else if constexpr (std::is_same_v) - return arg_a == arg_b; // identity comparison - else if constexpr (std::is_same_v) { - if (arg_a == arg_b) - return true; - if (!arg_a || !arg_b) - return false; - if (arg_a->size() != arg_b->size()) - return false; - if (!seen.insert({static_cast(arg_a.get()), - static_cast(arg_b.get())}) - .second) - return true; - for (std::size_t i = 0; i < arg_a->size(); ++i) - if ((*arg_a)[i].first != (*arg_b)[i].first) - return false; - for (std::size_t i = arg_a->size(); i-- > 0;) - pending.emplace_back(&(*arg_a)[i].second, &(*arg_b)[i].second); - return true; - } else if constexpr (std::is_same_v) - return false; // functions are never equal - else if constexpr (std::is_same_v) - return arg_a == arg_b; // identity comparison - else if constexpr (std::is_same_v) - return arg_a == arg_b; // identity comparison - else if constexpr (std::is_same_v) - return arg_a == arg_b; // byte-wise value comparison - else - return false; - }, - a.v); -} -} // namespace - -// The memo bounds work to the number of DISTINCT node-pairs (the physical size), never the -// exponential logical unfolding of a shared DAG, and terminates on cyclic structures; for -// finite acyclic values the result is identical to a naive compare. The pairs still to compare -// wait on a heap stack, so depth costs no native stack. A deadline poll bounds a huge -// NON-shared structure (memoization alone does not: distinct nodes are never re-encountered). -bool values_equal(const value_t &a, const value_t &b) { - ptr_pair_set seen; - value_pairs pending{{&a, &b}}; - while (!pending.empty()) { - if (eval_deadline_expired()) - throw std::runtime_error("values_equal: structure too large to compare"); - const auto [x, y] = pending.back(); - pending.pop_back(); - if (!shallow_equal(*x, *y, seen, pending)) - return false; - } - return true; -} +// to_string and values_equal are walks: see data_job.cpp. } // namespace cvc::state_exec diff --git a/src/cvc/tests/CMakeLists.txt b/src/cvc/tests/CMakeLists.txt index 46a6b65cd..7e99e1f04 100644 --- a/src/cvc/tests/CMakeLists.txt +++ b/src/cvc/tests/CMakeLists.txt @@ -80,6 +80,7 @@ add_executable(state_exec_integration_test state_exec_integration_test.cpp) add_executable(state_exec_advanced_integration_test state_exec_advanced_integration_test.cpp) add_executable(state_exec_gaps_test state_exec_gaps_test.cpp) add_executable(state_exec_deep_data_test state_exec_deep_data_test.cpp) +add_executable(state_exec_data_steps_test state_exec_data_steps_test.cpp) if(NOT WIN32) add_executable(state_exec_multiprocess_test state_exec_multiprocess_test.cpp) endif() @@ -361,6 +362,7 @@ list(APPEND TEST_TARGETS state_exec_advanced_integration_test state_exec_gaps_test state_exec_deep_data_test + state_exec_data_steps_test ) if(NOT WIN32) list(APPEND TEST_TARGETS state_exec_multiprocess_test) @@ -1242,6 +1244,13 @@ target_link_libraries(state_exec_deep_data_test GTest::gtest_main ) +target_link_libraries(state_exec_data_steps_test + PRIVATE + cvc + GTest::gtest + GTest::gtest_main +) + target_link_libraries(state_exec_builtins_test PRIVATE cvc @@ -1671,6 +1680,7 @@ target_compile_features(state_link_bench_test PRIVATE cxx_std_17) target_compile_features(state_exec_parser_test PRIVATE cxx_std_20) target_compile_features(state_exec_types_test PRIVATE cxx_std_20) target_compile_features(state_exec_deep_data_test PRIVATE cxx_std_20) +target_compile_features(state_exec_data_steps_test PRIVATE cxx_std_20) target_compile_features(state_exec_builtins_test PRIVATE cxx_std_20) target_compile_features(state_exec_evaluator_test PRIVATE cxx_std_20) target_compile_features(state_exec_stackless_evaluator_test PRIVATE cxx_std_20) @@ -2092,6 +2102,7 @@ cvc_discover_tests(state_link_bench_test) cvc_discover_tests(state_exec_parser_test) cvc_discover_tests(state_exec_types_test) cvc_discover_tests(state_exec_deep_data_test) +cvc_discover_tests(state_exec_data_steps_test) cvc_discover_tests(state_exec_builtins_test) cvc_discover_tests(state_exec_evaluator_test) cvc_discover_tests(state_exec_stackless_evaluator_test) diff --git a/src/cvc/tests/state_exec_bench_test.cpp b/src/cvc/tests/state_exec_bench_test.cpp index 52b1fef8f..f089768cd 100644 --- a/src/cvc/tests/state_exec_bench_test.cpp +++ b/src/cvc/tests/state_exec_bench_test.cpp @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -185,6 +186,50 @@ TEST_F(ExecBenchTest, ValueDropThroughput) { report("eval_nest_list_2000", M, std::chrono::duration_cast(t1 - t0)); } +TEST_F(ExecBenchTest, DataWalkThroughput) { + if (!bench_enabled()) { + SUCCEED(); + return; + } + + // Deep-data work per node, the unit of data.step_nodes: compare, print, copy, free. The value + // nests lists and dicts 20000 levels, each level holding a string and a number too. + constexpr int64_t kLevels = 20000; + const auto build = [] { + value_t v = value_t{int64_t{0}}; + for (int64_t i = 0; i < kLevels; ++i) + v = i % 2 ? make_dict({{"i", value_t{i}}, {"v", std::move(v)}}) + : make_list({value_t{std::string("s")}, std::move(v), value_t{1.5}}); + return v; + }; + const value_t a = build(), b = build(); + const auto per_node = [](const char *name, data_job &job) { + auto t0 = hrclock::now(); + job.step(0); + auto t1 = hrclock::now(); + report(name, static_cast(job.nodes()), std::chrono::duration_cast(t1 - t0)); + }; + auto eq = make_equal_job(a, b); + per_node("data_equal_per_node", *eq); + auto print = make_print_job(a); + per_node("data_print_per_node", *print); + auto copy = make_copy_job(a); + per_node("data_copy_per_node", *copy); + + // Freeing, through a release backlog drained whole. + value_t doomed = build(); + detail::release_backlog backlog; + { + detail::backlog_scope route(&backlog); + doomed = value_t{}; + } + auto t0 = hrclock::now(); + const std::size_t freed = backlog.drain(0); + auto t1 = hrclock::now(); + report("data_free_per_node", static_cast(freed), + 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 new file mode 100644 index 000000000..883dbc2b6 --- /dev/null +++ b/src/cvc/tests/state_exec_data_steps_test.cpp @@ -0,0 +1,446 @@ +// Deep-data work on the executor ("Deep data on the executor" in docs/STATE_EXEC_GRAMMAR.md): +// walks that run a slice at a time (data_job), the stackless evaluator slicing them +// (evaluator_state::data_step_nodes), the schedulers taking data.step_nodes from the state tree, +// and the budget split by kind (read / eval / data) that process_info, inspect and budget-usage +// report. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace cvc::state_exec; + +namespace { + +value_t sym(const char *s) { return value_t{symbol{s}}; } + +value_t nested_list(std::size_t n, value_t leaf = value_t{int64_t{1}}) { + value_t v = make_list({std::move(leaf)}); + for (std::size_t i = 1; i < n; ++i) + v = make_list({std::move(v)}); + return v; +} + +// Lists and dicts alternating, each level also holding a string and a number. +value_t nested_mixed(std::size_t n) { + value_t v = value_t{int64_t{0}}; + for (std::size_t i = 0; i < n; ++i) + v = i % 2 ? make_dict({{"i", value_t{int64_t(i)}}, {"v", std::move(v)}}) + : make_list({value_t{std::string("s")}, std::move(v), value_t{1.5}}); + return v; +} + +// Run `job` in slices of `slice` nodes. Returns its result and how many step() calls it took. +std::pair run_sliced(data_job &job, std::size_t slice) { + int calls = 1; + while (!job.step(slice)) + ++calls; + return {job.result(), calls}; +} + +// The script that builds two separate n-deep lists and compares them. +std::string compare_script(int n) { + return "(begin (set a 1) (set b 1) (set i 0)" + " (while (< i " + + std::to_string(n) + + ") (begin (set a (list a)) (set b (list b)) (set i (+ i 1))))" + " (= a b))"; +} + +// Step `st` until it is done; returns the steps it took. +int step_to_done(stackless_evaluator &ev, evaluator_state &st) { + int steps = 0; + while (!ev.step(st)) + ++steps; + return steps + 1; +} + +} // namespace + +// ---- the jobs --------------------------------------------------------------- + +TEST(DataJob, SlicedWalksMatchWholeOnes) { + const value_t a = nested_mixed(300), b = nested_mixed(300), shorter = nested_mixed(299); + for (std::size_t slice : {1u, 2u, 7u, 100000u}) { + SCOPED_TRACE("slice " + std::to_string(slice)); + auto eq = make_equal_job(a, b); + auto [equal, calls] = run_sliced(*eq, slice); + EXPECT_TRUE(std::get(equal.v)); + if (slice == 1) + EXPECT_GE(calls, 600); // a node per call + auto ne = make_equal_job(a, shorter); + EXPECT_FALSE(std::get(run_sliced(*ne, slice).first.v)); + + auto print = make_print_job(a); + EXPECT_EQ(std::get(run_sliced(*print, slice).first.v), to_string(a)); + + auto copy = make_copy_job(a); + value_t c = run_sliced(*copy, slice).first; + EXPECT_TRUE(values_equal(c, a)); + EXPECT_NE(std::get(c.v).get(), std::get(a.v).get()); // 300 levels: a dict + + auto concat = make_concat_job({value_t{std::string("<")}, a, value_t{int64_t{3}}}, " ", "test"); + EXPECT_EQ(std::get(run_sliced(*concat, slice).first.v), + "< " + to_string(a) + " 3"); + + const value_t tmpl = make_list({sym("x"), nested_list(50, sym("y")), sym("z")}); + auto sub = + make_substitute_job(tmpl, {{"x", value_t{int64_t{1}}}, {"y", value_t{std::string("Y")}}}); + EXPECT_TRUE(values_equal( + run_sliced(*sub, slice).first, + make_list({value_t{int64_t{1}}, nested_list(50, value_t{std::string("Y")}), sym("z")}))); + } +} + +TEST(DataJob, ConcatChecksTheAllocationLimit) { + // str-concat's sliced path (a list or dict among its parts) is held to the same allocation + // limit as its fast path: the same string passed many times can outgrow every part. + eval_alloc_guard limit(std::size_t{100}); + const value_t big{std::string(60, 'x')}; + auto job = make_concat_job({big, make_list({value_t{int64_t{1}}}), big}, "", "str-concat"); + try { + run_sliced(*job, 1); + FAIL() << "expected the allocation limit to stop it"; + } catch (const std::runtime_error &e) { + EXPECT_NE(std::string(e.what()).find("str-concat"), std::string::npos) << e.what(); + } +} + +TEST(DataJob, NodesDoNotDependOnTheSlice) { + const value_t v = nested_mixed(200); + auto whole = make_print_job(v); + whole->step(0); + auto sliced = make_print_job(v); + run_sliced(*sliced, 3); + EXPECT_EQ(whole->nodes(), sliced->nodes()); + EXPECT_GE(whole->nodes(), 200u); +} + +TEST(DataJob, CopyKeepsSharingAndCycles) { + value_t inner = make_list({value_t{int64_t{3}}}); + auto copy = make_copy_job(make_list({inner, inner})); + value_t c = run_to_completion(*copy); + auto &l = *std::get(c.v); + EXPECT_EQ(std::get(l[0].v).get(), std::get(l[1].v).get()); + EXPECT_NE(std::get(l[0].v).get(), std::get(inner.v).get()); + + auto cyclic = std::make_shared>(); + cyclic->push_back(value_t{cyclic}); + auto cc = make_copy_job(value_t{cyclic}); + value_t cy = run_sliced(*cc, 1).first; + auto copied = std::get(cy.v); + EXPECT_EQ(std::get((*copied)[0].v).get(), copied.get()); + copied->clear(); // break both cycles + cyclic->clear(); +} + +TEST(DataJob, ValueChangedBetweenSlices) { + // A job reads what it walks by index, holding it by owning pointer: a change between two + // slices (another process's) changes the answer, never memory safety. + list_ptr inner = std::make_shared>(); + for (int i = 0; i < 10; ++i) + inner->push_back(value_t{int64_t(i)}); + value_t a = make_list({value_t{inner}}); + value_t b = make_list({make_list(std::vector(*inner))}); + auto eq = make_equal_job(a, b); + EXPECT_FALSE(eq->step(3)); + inner->clear(); + inner->shrink_to_fit(); + EXPECT_FALSE(std::get(run_sliced(*eq, 3).first.v)); + + list_ptr printed = std::make_shared>(20, value_t{int64_t{7}}); + auto print = make_print_job(value_t{printed}); + EXPECT_FALSE(print->step(4)); + printed->clear(); + printed->shrink_to_fit(); + const std::string out = std::get(run_sliced(*print, 4).first.v); + EXPECT_EQ(out.front(), '('); + EXPECT_EQ(out.back(), ')'); +} + +TEST(DataJob, RunDataJobOutsideAStepRunsToCompletion) { + value_t r = run_data_job(make_equal_job(nested_list(100), nested_list(100))); + EXPECT_TRUE(std::get(r.v)); + r = run_data_job(make_print_job(nested_list(3)), + [](value_t s) { return value_t{std::get(s.v) + "!"}; }); + EXPECT_EQ(std::get(r.v), "(((1)))!"); +} + +// ---- the evaluator ------------------------------------------------------------ + +class DataStepsEvaluator : public ::testing::Test { +protected: + environment_ptr env = builtins::make_default_environment(); + stackless_evaluator ev{env}; + + // A state evaluating `script`, with data_step_nodes = `slice`, in an environment where `a` and + // `b` are separately built n-deep lists. + evaluator_state state(const std::string &script, std::size_t slice, std::size_t n = 1000) { + auto run_env = environment::extend(env); + run_env->set("a", nested_list(n)); + run_env->set("b", nested_list(n)); + evaluator_state st = ev.create_state(parse(script), run_env); + st.data_step_nodes = slice; + st.stats.start(); + return st; + } +}; + +TEST_F(DataStepsEvaluator, NativeCallRunsASlicePerStep) { + evaluator_state whole = state("(= a b)", 0); + const int whole_steps = step_to_done(ev, whole); + EXPECT_TRUE(std::get(whole.result.v)); + EXPECT_EQ(whole.stats.steps(step_kind::data), 0u); + EXPECT_GE(whole.stats.data_nodes(), 1000u); // counted, though done in one step + + evaluator_state sliced = state("(= a b)", 64); + const int sliced_steps = step_to_done(ev, sliced); + EXPECT_TRUE(std::get(sliced.result.v)); + EXPECT_GE(sliced.stats.steps(step_kind::data), 1000u / 64 - 1); + EXPECT_EQ(sliced_steps, whole_steps + static_cast(sliced.stats.steps(step_kind::data))); + EXPECT_EQ(sliced.stats.steps(step_kind::eval), whole.stats.steps(step_kind::eval)); +} + +TEST_F(DataStepsEvaluator, PrintingAndNotEqualSlice) { + evaluator_state s = state("(str a)", 50); + step_to_done(ev, s); + EXPECT_EQ(std::get(s.result.v), to_string(nested_list(1000))); + EXPECT_GT(s.stats.steps(step_kind::data), 10u); + + evaluator_state ne = state("(!= a b)", 50); + step_to_done(ev, ne); + EXPECT_FALSE(std::get(ne.result.v)); + EXPECT_GT(ne.stats.steps(step_kind::data), 10u); + + evaluator_state cat = state("(str-concat \"a=\" a)", 50); + step_to_done(ev, cat); + EXPECT_EQ(std::get(cat.result.v), "a=" + to_string(nested_list(1000))); +} + +TEST_F(DataStepsEvaluator, SmallWorkTakesNoExtraSteps) { + evaluator_state few = state("(= (list 1 2 3) (list 1 2 3))", 64); + evaluator_state whole = state("(= (list 1 2 3) (list 1 2 3))", 0); + EXPECT_EQ(step_to_done(ev, few), step_to_done(ev, whole)); + EXPECT_EQ(few.stats.steps(step_kind::data), 0u); +} + +TEST_F(DataStepsEvaluator, MacroExpansionSlices) { + const std::size_t n = 2000; + value_t expr = make_list({sym("begin"), + make_list({sym("defmacro"), sym("m"), make_list({sym("x")}), + make_list({sym("quote"), nested_list(n, sym("x"))})}), + make_list({sym("m"), value_t{int64_t{7}}})}); + evaluator_state st = ev.create_state(expr); + st.data_step_nodes = 64; + step_to_done(ev, st); + EXPECT_TRUE(values_equal(st.result, nested_list(n, value_t{int64_t{7}}))); + EXPECT_GE(st.stats.steps(step_kind::data), n / 64 - 1); +} + +TEST_F(DataStepsEvaluator, FreeingADeepValueSpreadsOverSteps) { + const std::size_t n = 10000; + auto run_env = environment::extend(env); + run_env->set("big", nested_list(n)); // its only owner + evaluator_state st = ev.create_state(parse("(begin (set big nil) 1)"), run_env); + st.data_step_nodes = 100; + st.stats.start(); + const int steps = step_to_done(ev, st); + EXPECT_EQ(std::get(st.result.v), 1); + EXPECT_GE(st.stats.steps(step_kind::data), n / 100 - 2); + EXPECT_GE(st.stats.data_nodes(), n); + // Evaluation waits while the backlog is freed, so the last step is an evaluation step. + EXPECT_EQ(static_cast(steps), st.stats.get_step_count()); +} + +TEST_F(DataStepsEvaluator, StepBudgetStopsDeepWorkMidway) { + evaluator_state st = state("(= a b)", 16); + ev.run(st, 8); + EXPECT_FALSE(st.done); + EXPECT_NE(st.data_work, nullptr); // waiting in the comparison + ev.run(st); + EXPECT_TRUE(st.done); + EXPECT_TRUE(std::get(st.result.v)); +} + +TEST_F(DataStepsEvaluator, SnapshotMidWorkRunsTheCallAgain) { + cvc::app ctx; + evaluator_state st = state("(= a b)", 16, 300); + ev.run(st, 8); + ASSERT_NE(st.data_work, nullptr); // in the middle of the comparison + + auto &node = cvc::state::instance(ctx)("snapshot"); + encode_evaluator_state(node, st); + evaluator_state back = decode_evaluator_state(node); + EXPECT_EQ(back.data_step_nodes, 16u); + EXPECT_EQ(back.data_work, nullptr); // work in flight is not kept... + // (native functions are not serialized: put the builtins back, as the codec tests do) + auto patch = [&](environment_ptr e) { + for (; e && e->outer; e = e->outer) { + } + if (e) + for (auto &[name, val] : env->bindings) + e->set(name, val); + }; + patch(back.global_env); + for (auto &f : back.stack) + patch(f.env); + ev.run(back); // ...the waiting call runs again + ASSERT_TRUE(back.done); + EXPECT_TRUE(std::get(back.result.v)); + EXPECT_GT(back.stats.steps(step_kind::data), 0u); +} + +// ---- the schedulers ------------------------------------------------------------- + +class DataStepsScheduler : public ::testing::Test { +protected: + cvc::app app_ctx; + cvc::state &root = cvc::state::instance(app_ctx); +}; + +TEST_F(DataStepsScheduler, SettingsComeFromTheTree) { + scheduler plain; + plain.set_watch_root(&root); + plain.set_id("plain"); + plain.load_settings(); + EXPECT_EQ(plain.data_settings().step_nodes, 0u); + EXPECT_EQ(root("state_exec.schedulers.plain.data.step_nodes").value(), "0"); // published + + root("state_exec.defaults.data.step_nodes").value("32"); + scheduler sched; + sched.set_watch_root(&root); + sched.set_id("ds"); + sched.load_settings(); + EXPECT_EQ(sched.data_settings().step_nodes, 32u); + EXPECT_EQ(root("state_exec.schedulers.ds.data.step_nodes").value(), "32"); + root("state_exec.schedulers.ds.data.step_nodes").value("8"); + sched.load_settings(); + EXPECT_EQ(sched.data_settings().step_nodes, 8u); + root("state_exec.schedulers.ds.data.step_nodes").value("lots"); // skipped + sched.load_settings(); + EXPECT_EQ(sched.data_settings().step_nodes, 32u); + + async_scheduler async; + EXPECT_EQ(async.data_settings().step_nodes, 0u); + async.set_data_defaults(async_scheduler::host_data_defaults()); + EXPECT_EQ(async.data_settings().step_nodes, async_scheduler::host_data_defaults().step_nodes); + async.set_watch_root(&root); // state_exec.defaults.data.step_nodes = 32 overrides it + EXPECT_EQ(async.data_settings().step_nodes, 32u); +} + +TEST_F(DataStepsScheduler, DeepWorkCountsAgainstMaxSteps) { + root("state_exec.defaults.data.step_nodes").value("16"); + scheduler sched; + sched.set_watch_root(&root); + sched.load_settings(); + + const int done = sched.execute(compare_script(500)); + sched.run(); + auto info = sched.get_process_info(done); + ASSERT_TRUE(info); + EXPECT_EQ(info->status, process_status::terminated); + EXPECT_TRUE(std::get(sched.get_result(done)->v)); + EXPECT_GE(info->data_steps, 500u / 16 - 1); + EXPECT_GT(info->eval_steps, 0u); + EXPECT_EQ(info->read_steps, 0u); + EXPECT_GE(info->data_nodes, 500u); + EXPECT_EQ(info->eval_steps + info->data_steps + info->read_steps, info->step_count); + EXPECT_GT(info->eval_time, 0.0); + EXPECT_GT(info->data_time, 0.0); + + // The same program with room for its evaluation but not its comparison. + execute_options opts; + opts.max_steps = info->eval_steps + 2; + const int killed = sched.execute(compare_script(500), opts); + sched.run(); + EXPECT_EQ(sched.get_process_info(killed)->status, process_status::killed); +} + +TEST_F(DataStepsScheduler, OtherProcessesRunBetweenSlices) { + root("state_exec.defaults.data.step_nodes").value("4"); + scheduler sched; + sched.set_watch_root(&root); + sched.load_settings(); + const int slow = sched.execute(compare_script(400)); + // Let it reach its comparison. + while (sched.get_process_info(slow)->data_steps == 0) + sched.step(); + const int quick = sched.execute(std::string("(begin 1 2 3)")); + while (sched.get_process_info(quick)->status != process_status::terminated) + sched.step(); + EXPECT_NE(sched.get_process_info(slow)->status, process_status::terminated); + sched.run(); + EXPECT_TRUE(std::get(sched.get_result(slow)->v)); +} + +TEST_F(DataStepsScheduler, SignalsWaitForDeepWork) { + root("state_exec.defaults.data.step_nodes").value("4"); + scheduler sched; + sched.set_watch_root(&root); + sched.load_settings(); + execute_options opts; + opts.signal_handlers["SIGUSR1"] = make_list({sym("begin"), value_t{int64_t{42}}}); + const int pid = sched.execute(compare_script(200), opts); + while (sched.get_process_info(pid)->data_steps == 0) + sched.step(); + ASSERT_TRUE(sched.send_signal(pid, "SIGUSR1")); + const uint64_t before = sched.get_process_info(pid)->data_steps; + sched.step(); + EXPECT_EQ(sched.get_process_info(pid)->data_steps, before + 1); // the work went on + sched.run(); + EXPECT_EQ(sched.get_process_info(pid)->status, process_status::terminated); + EXPECT_TRUE(std::get(sched.get_result(pid)->v)); +} + +TEST_F(DataStepsScheduler, BudgetUsageAndInspectReportTheKinds) { + async_scheduler sched; + sched.set_watch_root(&root); + root("state_exec.defaults.data.step_nodes").value("16"); + memory_tracker tracker; + intrinsics_context ictx; + ictx.sched = &sched; + ictx.root = &root; + ictx.tracker = &tracker; + auto env = builtins::make_default_environment(); + register_intrinsics(env, &ictx); + execute_options opts; + opts.env = env; + const int pid = sched.execute( + std::string("(begin (set a 1) (set b 1) (set i 0)" + " (while (< i 300) (begin (set a (list a)) (set b (list b)) (set i (+ i 1))))" + " (= a b) (set used (budget-usage)) (set me (inspect (self-pid)))" + " (list used me))"), + opts); + sched.sync_run(); + auto result = sched.get_result(pid); + ASSERT_TRUE(result); + auto &pair = *std::get(result->v); + for (const value_t &d : pair) { + std::unordered_map fields; + for (auto &[k, v] : *std::get(d.v)) + fields[k] = v; + for (const char *key : {"read_steps", "eval_steps", "data_steps", "data_nodes", "read_time", + "eval_time", "data_time", "steps"}) + EXPECT_TRUE(fields.count(key)) << key; + EXPECT_GE(std::get(fields["data_steps"].v), 300 / 16 - 1); + EXPECT_GE(std::get(fields["data_nodes"].v), 300); + EXPECT_GT(std::get(fields["eval_steps"].v), 0); + } + auto info = sched.get_process_info(pid); + ASSERT_TRUE(info); + EXPECT_GT(info->data_time, 0.0); +}