diff --git a/bindings/pycvc/test_pycvc_exec.py b/bindings/pycvc/test_pycvc_exec.py index 244fbb6ef..d624657f3 100644 --- a/bindings/pycvc/test_pycvc_exec.py +++ b/bindings/pycvc/test_pycvc_exec.py @@ -335,6 +335,23 @@ def run_error(src): assert ex.run("(length (collections.range 1000))") == "1000" +# ── reference cycles ──────────────────────────────────────────────────── + + +def test_exec_functions_survive_cycle_collection(): + app = pycvc.make_app() + ex = pycvc.Exec(app) + ex.run("(defun twice (x) (* 2 x))") + # Each iteration leaks a let scope (a closure bound in the scope it captured). That many make + # the scheduler collect cycles between steps, and Exec's own environment, held by Exec, must + # come through intact. + ex.run( + "(begin (set i 0) (while (< i 1000) (begin (let ((h nil)) (set h (lambda () h))) " + "(set i (+ i 1)))) i)" + ) + assert ex.run("(twice 21)") == "42" + + if __name__ == "__main__": tests = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)] for t in tests: diff --git a/docs/ARIADNE.md b/docs/ARIADNE.md index 90da9dc59..24d5598bb 100644 --- a/docs/ARIADNE.md +++ b/docs/ARIADNE.md @@ -128,9 +128,12 @@ per-activation caps ([`ariadne.cpp:1382`](../src/cvc/ariadne/ariadne.cpp)): | wall-time cap | `kActionMaxSeconds` | `0.1 s` | | memory budget | tree setting `ariadne.limits.action_max_memory` | `4 MiB` by default; `0` = no limit | -The memory budget is read on every submission. It bounds the action's state-tree writes and -also each native call's allocations: state_exec's per-step allocation limit takes what is left -of `max_memory` (developer guide §11, "Allocation limit"). +The memory budget is read on every submission. It covers the action's state-tree writes and its +measured share of the values it holds (developer guide §7), and it bounds each native call's +allocations: state_exec's per-step allocation limit takes what is left of `max_memory` +(developer guide §11, "Allocation limit"). A finished action's process is reaped from the app +scheduler when the drain sweeps it, and what it left in reference cycles (a function it defined, +say) is freed by the cycle collector. A quick action completes within that drain. An action that `await`s, `sleep`s, or `msg-recv`s **parks** and resumes on a later frame — this is what makes the async arc diff --git a/docs/STATE_EXEC_DEVELOPER_GUIDE.md b/docs/STATE_EXEC_DEVELOPER_GUIDE.md index 393a689a9..296c3563d 100644 --- a/docs/STATE_EXEC_DEVELOPER_GUIDE.md +++ b/docs/STATE_EXEC_DEVELOPER_GUIDE.md @@ -819,13 +819,70 @@ Set via `execute_options` at submission time or adjusted at runtime: execute_options opts; opts.max_steps = 10000; // evaluator steps opts.max_time = 5.0; // wall time in seconds -opts.max_memory = 1048576; // bytes (tracked by memory_tracker) +opts.max_memory = 1048576; // bytes: state-tree writes + the measured heap share opts.max_messages = 100; // outbound msg-send calls opts.max_message_bytes = 65536; // total outbound payload bytes ``` When a limit is exceeded, the process is killed (status becomes `killed`). +### Memory: state-tree writes and the measured heap + +`max_memory` covers two things: +- what the process wrote to the state tree, counted by the scheduler's `memory_tracker`; +- its share of the values it holds in memory (lists, dicts, strings, closures, environments), + measured by walking what it can reach (`heap.h`). + +A value can be shared between processes. A broadcast message is one dict held by every +receiver, and a forked child shares its parent's lists. So a process pays its **share** of each +node it reaches: the node's size × (its references ÷ the node's owner count), and a node passes +its share on to the nodes it holds. Summed over all holders, a node's shares come to its size: a +1 MB message held by 50 residents costs each about 20 KB, and as holders drop it, those left pay +more at their next measurement. + +A measurement is the process's own deep-data work, run a slice per step like other data work +(`data.step_nodes`) and counted against its steps and time. One starts after a step when any of +these holds: + +- natives have reported allocating 64 KiB, or half the last measurement if that is more, since + the last one (the functions with an allocation check, §11, plus copies like `cons` and + `slice`); +- what they reported could cross the remaining budget; +- `limits.heap_check_steps` steps have passed (default 4096; `0` = only the two above). + +The kill decision always uses an actual measurement, never the reported bytes, so temporaries a +program drops don't count. Only a process with `max_memory > 0` is measured. +`process_info::heap_memory` and `heap_measurements` report the last measurement. The per-step +allocation limit (§11) takes what is left after state-tree writes and the measured heap. + +### Reference cycles + +Values are reference counted, and reference counting can't free a cycle. These all leaked their +whole scope, with everything in it, before the collector existed: +- a closure bound in the scope it captured: every `defun`, `(set f (lambda …))`, a `let`-local + recursive lambda; +- a class, whose methods capture the scope the class is bound in; +- a container that holds itself, as after `(append l l)`. + +`collect_cycles()` (`heap.h`) finds the cycles nothing outside them holds, the way CPython does: +- It tracks the environments that closures, generators and class methods capture, and the + containers changed in place to hold another container (`append`, `set-nth`, `set-attr`). Only + those can close a cycle. +- It counts the references these objects (and everything they reach) make to each other. Anything + with more owners than that is held from outside: by a live process, a host (pycvc's `Exec`, an + Ariadne action context), or anything else. Such an object stays, along with everything it + reaches. +- What is left is garbage. Clearing it lets it free. + +Both schedulers collect between steps once a thread has tracked as many new objects as survived +the last collection, and at least an eighth as many as it walked (at least 256). That keeps a +collection's cost proportional to what was tracked since the last one. They also collect at most +a second after a process ended. A finished +process also releases its evaluation state right away (`process::release_state`; result, error +and statistics stay), so what it leaked can be collected before the process is reaped. Tracking +is per thread. A host that runs evaluators without a scheduler calls `collect_cycles()` itself, +between steps; inside a step it does nothing. + ### Cluster-Level Resource Policy A `resource_policy` defines cluster-wide constraints that apply to all @@ -890,6 +947,7 @@ sched.load_settings(); // read and apply settings |---------|-------------------|-------------|---------| | `max_pending_messages` | `state_exec.schedulers..max_pending_messages` | `state_exec.defaults.max_pending_messages` | `1024` | | `limits.max_alloc_bytes` | `state_exec.schedulers..limits.max_alloc_bytes` | `state_exec.defaults.limits.max_alloc_bytes` | `16777216` ([§11](#allocation-limit)) | +| `limits.heap_check_steps` | `state_exec.schedulers..limits.heap_check_steps` | `state_exec.defaults.limits.heap_check_steps` | `4096` ([heap](#memory-state-tree-writes-and-the-measured-heap)) | ```cpp // Set a global default for all schedulers @@ -910,6 +968,7 @@ configuration back to the state tree: - `state_exec.schedulers..max_pending_messages` — the resolved cap - `state_exec.schedulers..limits.max_alloc_bytes` — the resolved per-call allocation limit +- `state_exec.schedulers..limits.heap_check_steps` — the resolved heap re-measure interval - `state_exec.schedulers..policy` — `"round_robin"`, `"priority"`, or `"priority_rr"` --- @@ -1341,8 +1400,8 @@ mine.import_module("geo", env, {"geo.area"}); // or a subset ### Allocation limit A native call builds its whole result inside one evaluator step. The per-process -`max_memory` check runs between steps and counts only state-tree writes, so it cannot see -`(collections.range 1000000000)`. Each function whose result can be larger than all of its +`max_memory` check runs between steps, and the heap it counts is measured after the fact (§7), so +neither can stop `(collections.range 1000000000)` from allocating. Each function whose result can be larger than all of its arguments therefore estimates the result's size and fails before allocating when it is over the limit: @@ -1353,7 +1412,8 @@ the limit: The limit for a step is the tighter of two values: 1. the tree setting `limits.max_alloc_bytes` (table below; `0` means no limit of its own); -2. what is left of the process's `max_memory`, when it has one. +2. what is left of the process's `max_memory`, when it has one: after its state-tree writes and + its last measured heap share (§7). | Setting | Per-scheduler path | Global path | Default | |---------|-------------------|-------------|---------| diff --git a/inc/cvc/state/state_exec/async_scheduler.h b/inc/cvc/state/state_exec/async_scheduler.h index 9c2231a93..7d09f264d 100644 --- a/inc/cvc/state/state_exec/async_scheduler.h +++ b/inc/cvc/state/state_exec/async_scheduler.h @@ -215,6 +215,7 @@ class async_scheduler : public scheduler_base { uint64_t total_steps_ = 0; int current_pid_ = -1; + uint32_t cycle_check_ = 0; // steps, for checking every 64th whether a cycle collection is due process_ptr current_proc_; // The scheduled evaluator IS the async_stackless_evaluator (roadmap §8.12): a coroutine @@ -246,6 +247,9 @@ class async_scheduler : public scheduler_base { void check_limits(process &proc); void terminate_process(process &proc, value_t result); void kill_process(process &proc, const std::string &reason); + /// After a process ended: note it for the cycle collector and, unless it is the one stepping + /// now, release its evaluation state (process::release_state). + void release_if_idle(process &proc); process_info make_info(const process &proc) const; }; diff --git a/inc/cvc/state/state_exec/heap.h b/inc/cvc/state/state_exec/heap.h new file mode 100644 index 000000000..93eeca91a --- /dev/null +++ b/inc/cvc/state/state_exec/heap.h @@ -0,0 +1,110 @@ +/** + * @file heap.h + * @brief A process's live DSL heap: its share for max_memory, and collecting reference cycles. + * + * Values are reference counted and may be shared between processes: a broadcast message is one + * dict held by every receiver, and a forked child shares its parent's lists. A process is charged + * its share of each node it can reach, size x (its references / the node's owner count), and a + * node passes its share on to the nodes under it. A node's shares over all its holders add up to + * its size, so a value held by k processes costs each of them about 1/k of it. + * + * Reference counting cannot free a cycle: a closure bound in the environment it captured (every + * defun), or a list appended to itself. The collector here finds the cycles nothing outside them + * holds, as CPython's does: it counts the references the tracked objects make to each other, and + * whatever has more owners than that is held from outside, as is everything it reaches. The rest + * is garbage, and clearing it lets it free. + */ +#ifndef CVC_STATE_EXEC_HEAP_H +#define CVC_STATE_EXEC_HEAP_H + +#include +#include +#include +#include +#include + +namespace cvc { +class state; +} + +namespace cvc::state_exec { + +class data_job; +struct process; + +// --------------------------------------------------------------------------- +// Measuring a process's heap +// --------------------------------------------------------------------------- + +/// Heap settings. Hosts keep them on the state tree under `limits.*`, like +/// limits.max_alloc_bytes. +struct heap_options { + /// Re-measure a process that has a memory budget at least every this many steps, besides the + /// measurements its allocations call for (0 = only those). State-tree key: + /// `limits.heap_check_steps`. + std::size_t check_steps = 4096; +}; + +/// Resolve heap settings from a state tree, per key: `state_exec.schedulers..limits.`, +/// then `state_exec.defaults.limits.`, then `base`. A missing or malformed value is skipped. +heap_options heap_options_from_state(cvc::state &root, const std::string &scheduler_id = {}, + const heap_options &base = {}); + +/// A data job that measures `proc`'s heap share; its result is the share in bytes, as an int. +/// It holds what it measures, so the process (and others) may run between its slices. +std::unique_ptr make_heap_share_job(const process &proc); + +/// `proc`'s heap share in bytes, measured now in one go. +std::size_t measure_heap_share(const process &proc); + +/// The bytes a re-measure waits for: natives' reported allocations since the last measurement +/// trigger one once they reach this (64 KiB, or half the last measurement if that is more). +std::size_t heap_remeasure_bytes(std::size_t last_measured); + +/// The heap accounting a scheduler does after each step of `proc`: take a measurement that just +/// finished (proc.heap_bytes), add `allocated` (the step guard's eval_alloc_guard::allocated()), +/// and start the next measurement as the process's data work when one is due +/// (process::heap_measure_due). `state_bytes` is the memory tracker's count for the process. +/// Nothing is measured for a process without a memory budget. +void heap_after_step(process &proc, std::size_t allocated, uint64_t state_bytes); + +// --------------------------------------------------------------------------- +// Reference cycles +// --------------------------------------------------------------------------- + +/// Track an environment that a closure, a generator or a class method captured: it may close a +/// cycle (the closure bound in it, or in something it holds). Tracking is per thread. +void track_captured_env(const environment_ptr &env); + +/// Track a list or dict that was changed in place (append, set-nth, set-attr): only a change in +/// place can make a container hold itself. +void track_mutated(const list_ptr &list); +void track_mutated(const dict_ptr &dict); + +/// What a collection did. +struct cycle_stats { + std::size_t candidates = 0; ///< tracked objects still alive when it started + std::size_t examined = 0; ///< objects reachable from them + std::size_t freed = 0; ///< objects found to be garbage and cleared +}; + +/// Find the reference cycles among this thread's tracked objects that nothing outside them holds, +/// and clear them so they free. Does nothing (and returns zeros) while an evaluator step is +/// running on this thread: call it between steps. The schedulers call it when one is due. +cycle_stats collect_cycles(); + +/// Whether this thread has tracked enough since its last collection for one to be due: as many +/// new objects as survived the last one, at least an eighth of the objects it walked (so a walk +/// costs a bounded amount per newly tracked object), and at least 256. +bool cycle_collection_due(); + +/// Note that a process ended on this thread: what it leaked is garbage now, so a collection is +/// due as soon as anything has been tracked since the last one (at most once a second). +void note_process_exit(); + +/// Objects this thread tracks: those alive after its last collection plus those tracked since. +std::size_t tracked_objects(); + +} // namespace cvc::state_exec + +#endif // CVC_STATE_EXEC_HEAP_H diff --git a/inc/cvc/state/state_exec/process.h b/inc/cvc/state/state_exec/process.h index 39053c693..2688c34ea 100644 --- a/inc/cvc/state/state_exec/process.h +++ b/inc/cvc/state/state_exec/process.h @@ -73,6 +73,17 @@ struct process { // limits.max_alloc_bytes, resolved when the process is submitted. See alloc_limit(). uint64_t max_alloc_bytes = kDefaultMaxAllocBytes; + // Heap accounting (heap.h), for a process with max_memory > 0: max_memory covers its state-tree + // writes plus heap_bytes, its measured share of the values it can reach. A measurement runs as + // the process's own data steps; heap_alloc_bytes (what natives reported allocating) and + // heap_steps since the last one decide when the next is due. + uint64_t heap_bytes = 0; + uint64_t heap_alloc_bytes = 0; + uint64_t heap_steps = 0; + uint64_t heap_measurements = 0; + std::size_t heap_check_steps = 4096; // limits.heap_check_steps, resolved at submission + bool heap_measuring = false; // a measurement is in proc.state.data_work + // Evaluator state (stackless — serializable) evaluator_state state; @@ -147,11 +158,23 @@ struct process { /// The allocation limit for the next step's eval_alloc_guard (nullopt when unlimited): the /// tighter of max_alloc_bytes and what is left of max_memory once `used_bytes` (the memory - /// tracker's count for this process) is taken out. + /// tracker's count for this process plus heap_bytes) is taken out. std::optional alloc_limit(std::size_t used_bytes) const; + /// Whether to start a heap measurement after this step: the process has a memory budget, none + /// is running, and natives reported enough allocation (heap_remeasure_bytes), or the next + /// allocation could cross the budget, or limits.heap_check_steps steps passed. + /// `state_bytes` is the memory tracker's count for this process. + bool heap_measure_due(uint64_t state_bytes) const; + /// Step count from evaluator stats. uint64_t step_count() const { return state.stats.get_step_count(); } + + /// Drop what a finished process evaluated with (its stack, environment, program, handlers and + /// inbox), keeping its result, error and statistics. A scheduler calls it once the process is + /// terminated or killed and not running, so what the program left in reference cycles becomes + /// garbage the cycle collector can free (heap.h), even before the process is reaped. + void release_state(); }; /// Shared ownership alias for process. diff --git a/inc/cvc/state/state_exec/scheduler.h b/inc/cvc/state/state_exec/scheduler.h index 0401c6245..1d8827c38 100644 --- a/inc/cvc/state/state_exec/scheduler.h +++ b/inc/cvc/state/state_exec/scheduler.h @@ -11,6 +11,7 @@ #define CVC_STATE_EXEC_SCHEDULER_H #include +#include #include #include #include @@ -210,6 +211,8 @@ class scheduler : public scheduler_base { scheduling_policy policy_; parse_options parse_options_; data_options data_options_; + heap_options heap_options_; + uint32_t cycle_check_ = 0; // steps, for checking every 64th whether a cycle collection is due int next_pid_ = 1; int rr_index_ = 0; bool running_ = false; @@ -257,6 +260,10 @@ class scheduler : public scheduler_base { /// Mark a process as killed with an error. void kill_process(process &proc, const std::string &reason); + /// After a process ended: note it for the cycle collector and, unless it is the one stepping + /// now, release its evaluation state (process::release_state). + void release_if_idle(process &proc); + /// Build process_info from a process. process_info make_info(const process &proc) const; }; diff --git a/inc/cvc/state/state_exec/scheduler_base.h b/inc/cvc/state/state_exec/scheduler_base.h index f17e6371a..6eda0ee9d 100644 --- a/inc/cvc/state/state_exec/scheduler_base.h +++ b/inc/cvc/state/state_exec/scheduler_base.h @@ -86,6 +86,8 @@ struct process_info { double elapsed_time; uint64_t current_memory; uint64_t peak_memory; + uint64_t heap_memory = 0; // measured heap share (heap.h); max_memory covers it too + uint64_t heap_measurements = 0; // measurements taken so far uint64_t max_memory; double max_time; uint64_t message_count; diff --git a/inc/cvc/state/state_exec/stackless_evaluator.h b/inc/cvc/state/state_exec/stackless_evaluator.h index bcd88d470..b1adca36b 100644 --- a/inc/cvc/state/state_exec/stackless_evaluator.h +++ b/inc/cvc/state/state_exec/stackless_evaluator.h @@ -66,10 +66,12 @@ struct eval_frame { }; /// 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). +/// frame waits in the apply phase), the expansion of a macro call (its frame waits to be +/// re-initialized with it), or a measurement of the process's heap that its scheduler started +/// between two steps (heap.h; its result goes to evaluator_state::heap_measured, and evaluation +/// resumes where it was). struct pending_data_work { - enum class kind { native_call, macro_expansion }; + enum class kind { native_call, macro_expansion, heap_measure }; std::unique_ptr job; std::function then; // native_call: turns the job's result into the call's kind what = kind::native_call; @@ -99,6 +101,9 @@ struct evaluator_state { /// 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; + /// A heap measurement's result in bytes, once its data steps finish; the scheduler that started + /// it takes it from here. Not serialized. + std::optional heap_measured; }; /// Synchronous stackless evaluator (Tier 2 — local contexts only). @@ -115,6 +120,9 @@ class stackless_evaluator { public: explicit stackless_evaluator(environment_ptr global_env); + /// The global environment new states extend (the cycle collector walks it from a generator). + const environment_ptr &global_env() const { return global_env_; } + /// Create a fresh evaluator_state from a parsed expression. evaluator_state create_state(const value_t &expr, environment_ptr env = nullptr); diff --git a/inc/cvc/state/state_exec/types.h b/inc/cvc/state/state_exec/types.h index 4a2e76e9d..e3a14c2f1 100644 --- a/inc/cvc/state/state_exec/types.h +++ b/inc/cvc/state/state_exec/types.h @@ -406,18 +406,31 @@ class eval_alloc_guard { eval_alloc_guard(const eval_alloc_guard &) = delete; eval_alloc_guard &operator=(const eval_alloc_guard &) = delete; + /// Bytes natives allocated while this guard was the innermost one: every eval_alloc_check() + /// that passed, and every eval_alloc_note(). A scheduler adds them to the process's heap + /// accounting (heap.h), which uses them to decide when to re-measure. + std::size_t allocated() const { return allocated_; } + private: friend std::optional eval_alloc_limit(); + friend void eval_alloc_check(std::size_t, const char *); + friend void eval_alloc_note(std::size_t); std::optional limit_; // the effective limit: this guard's, tightened by outer ones eval_alloc_guard *outer_ = nullptr; + std::size_t allocated_ = 0; }; /// The allocation limit in force on this thread (std::nullopt: unbounded). std::optional eval_alloc_limit(); -/// Throw std::runtime_error naming `who` when `bytes` is over eval_alloc_limit(). +/// Throw std::runtime_error naming `who` when `bytes` is over eval_alloc_limit(); otherwise +/// report the bytes as allocated (eval_alloc_guard::allocated()). void eval_alloc_check(std::size_t bytes, const char *who); +/// Report `bytes` a native allocated without a limit check (a copy no bigger than an argument, +/// such as cons or slice), for heap accounting only. +void eval_alloc_note(std::size_t bytes); + /// `count * each`, saturating at SIZE_MAX, for size estimates that must not wrap. inline std::size_t alloc_bytes(std::size_t count, std::size_t each) { return each != 0 && count > SIZE_MAX / each ? SIZE_MAX : count * each; diff --git a/src/cvc/CMakeLists.txt b/src/cvc/CMakeLists.txt index 8c440805a..32a091091 100644 --- a/src/cvc/CMakeLists.txt +++ b/src/cvc/CMakeLists.txt @@ -88,6 +88,7 @@ set(INCLUDE_FILES ../../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/heap.h ../../inc/cvc/state/state_exec/process.h ../../inc/cvc/state/state_exec/memory_tracker.h ../../inc/cvc/state/state_exec/scheduler_base.h @@ -211,6 +212,8 @@ set(SOURCE_FILES state/state_exec/async_stackless_evaluator.cpp state/state_exec/state_value_codec.cpp state/state_exec/data_job.cpp + state/state_exec/heap.cpp + state/state_exec/closure_natives.h state/state_exec/process.cpp state/state_exec/memory_tracker.cpp state/state_exec/scheduler.cpp diff --git a/src/cvc/ariadne/ariadne.cpp b/src/cvc/ariadne/ariadne.cpp index dc725bcba..ef99a69cf 100644 --- a/src/cvc/ariadne/ariadne.cpp +++ b/src/cvc/ariadne/ariadne.cpp @@ -1492,12 +1492,14 @@ void Runtime::Impl::pump_and_sweep_actions() { for (auto it = live_actions_.begin(); it != live_actions_.end();) { auto info = sched.get_process_info(it->first); if (!info || info->status == se::process_status::terminated) { + sched.reap(it->first); // nothing reads a finished action's result: free its process it = live_actions_.erase(it); // normal completion (or already reaped from the process map) } else if (info->status == se::process_status::killed) { // The exit error says why: a syntax error (scripts are read inside the process), a runtime // error or a resource limit. warn_once("ari: on: action program did not complete: " + sched.get_exit_error(it->first).value_or("killed")); + sched.reap(it->first); it = live_actions_.erase(it); } else { ++it; // ready/running/waiting — still in flight (e.g. suspended on await/msg-recv/sleep) diff --git a/src/cvc/state/state_exec/async_scheduler.cpp b/src/cvc/state/state_exec/async_scheduler.cpp index 578f40fec..7e9b2c625 100644 --- a/src/cvc/state/state_exec/async_scheduler.cpp +++ b/src/cvc/state/state_exec/async_scheduler.cpp @@ -38,6 +38,8 @@ int async_scheduler::execute(const std::string &script, const execute_options &o // No load_settings() here: read the allocation limit from the tree on every submission. proc.max_alloc_bytes = watch_root_ ? max_alloc_bytes_from_state(*watch_root_, {}) : kDefaultMaxAllocBytes; + proc.heap_check_steps = + (watch_root_ ? heap_options_from_state(*watch_root_) : heap_options{}).check_steps; proc.signal_handlers = opts.signal_handlers; proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); @@ -72,6 +74,8 @@ int async_scheduler::execute(const value_t &expr, const execute_options &opts) { // No load_settings() here: read the allocation limit from the tree on every submission. proc.max_alloc_bytes = watch_root_ ? max_alloc_bytes_from_state(*watch_root_, {}) : kDefaultMaxAllocBytes; + proc.heap_check_steps = + (watch_root_ ? heap_options_from_state(*watch_root_) : heap_options{}).check_steps; proc.signal_handlers = opts.signal_handlers; proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); @@ -137,6 +141,11 @@ process_ptr async_scheduler::select_process() { // --------------------------------------------------------------------------- void async_scheduler::execute_process_step(process &proc) { + // Between steps: free the reference cycles programs leaked, once enough piled up (heap.h). + // Checked every 64 steps: the check reads thread-local state. + if ((++cycle_check_ & 63) == 0 && cycle_collection_due()) + collect_cycles(); + // Track which process is currently executing so intrinsics can query it (self/ps). current_pid_ = proc.pid; current_proc_ = processes_[proc.pid]; @@ -148,6 +157,14 @@ void async_scheduler::execute_process_step(process &proc) { pp.reset(); } } reset_guard{current_pid_, current_proc_}; + // Once this step is over, a process it finished lets go of what it evaluated with. + struct release_guard { + process &p; + ~release_guard() { + if (p.status == process_status::terminated || p.status == process_status::killed) + p.release_state(); + } + } release_dead{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. One @@ -185,8 +202,8 @@ void async_scheduler::execute_process_step(process &proc) { // Armed only when it changes something (see scheduler.cpp). std::optional step_alloc; if (proc.max_memory > 0 || proc.max_alloc_bytes != kDefaultMaxAllocBytes) - step_alloc.emplace( - proc.alloc_limit(proc.max_memory > 0 ? mem_tracker_.current_bytes(proc.pid) : 0)); + step_alloc.emplace(proc.alloc_limit( + proc.max_memory > 0 ? mem_tracker_.current_bytes(proc.pid) + proc.heap_bytes : 0)); const bool charge_cpu = proc.charges_cpu_time(); const double cpu0 = charge_cpu ? thread_cpu_seconds() : 0.0; // Drive one async_stackless_evaluator step to completion at the leaf. The evaluator's step() @@ -222,6 +239,10 @@ void async_scheduler::execute_process_step(process &proc) { 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()); + // Heap accounting: take a finished measurement, start the next when due (heap.h). + if (proc.max_memory > 0 || proc.state.heap_measured) + heap_after_step(proc, step_alloc ? step_alloc->allocated() : 0, + mem_tracker_.current_bytes(proc.pid)); if ((proc.in_signal_handler || proc.in_watch_handler) && proc.state.done) { restore_from_signal(proc); @@ -612,7 +633,9 @@ void async_scheduler::check_limits(process &proc) { kill_process(proc, "time_limit_exceeded"); return; } - if (proc.max_memory > 0 && mem_tracker_.current_bytes(proc.pid) >= proc.max_memory) { + // max_memory: state-tree writes plus the measured heap share (heap.h) + if (proc.max_memory > 0 && + mem_tracker_.current_bytes(proc.pid) + proc.heap_bytes >= proc.max_memory) { kill_process(proc, "memory_limit_exceeded"); return; } @@ -637,12 +660,21 @@ void async_scheduler::terminate_process(process &proc, value_t result) { if (proc.on_complete) { proc.on_complete(proc.exit_code); } + release_if_idle(proc); } void async_scheduler::kill_process(process &proc, const std::string &reason) { proc.status = process_status::killed; proc.exit_error = reason; proc.state.stats.mark_complete(); + release_if_idle(proc); +} + +void async_scheduler::release_if_idle(process &proc) { + note_process_exit(); // what it leaked is garbage now (heap.h) + // A process stepping right now (one that killed itself) is released once its step is over. + if (proc.pid != current_pid_) + proc.release_state(); } bool async_scheduler::pause(int pid) { @@ -716,6 +748,7 @@ int async_scheduler::fork(int pid) { child->max_step_time = parent.max_step_time; child->max_memory = parent.max_memory; child->max_alloc_bytes = parent.max_alloc_bytes; + child->heap_check_steps = parent.heap_check_steps; child->max_messages = parent.max_messages; child->max_message_bytes = parent.max_message_bytes; child->signal_handlers = parent.signal_handlers; @@ -812,6 +845,8 @@ process_info async_scheduler::make_info(const process &proc) const { info.elapsed_time = proc.elapsed_time(); info.current_memory = mem_tracker_.current_bytes(proc.pid); info.peak_memory = mem_tracker_.peak_bytes(proc.pid); + info.heap_memory = proc.heap_bytes; + info.heap_measurements = proc.heap_measurements; info.max_memory = proc.max_memory; info.max_time = proc.max_time; info.message_count = proc.message_count; diff --git a/src/cvc/state/state_exec/builtins.cpp b/src/cvc/state/state_exec/builtins.cpp index 3491f9aca..a55bce6cd 100644 --- a/src/cvc/state/state_exec/builtins.cpp +++ b/src/cvc/state/state_exec/builtins.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include // SE-1: (length "...") counts codepoints for a string #include @@ -29,6 +30,14 @@ double as_double(const value_t &v) { /// True when the value is an integer. bool is_int(const value_t &v) { return std::holds_alternative(v.v); } +/// True when the value can hold other values, so storing it in a container in place may close a +/// reference cycle (heap.h): the cycle collector then tracks that container. +bool can_close_cycle(const value_t &v) { + return std::holds_alternative(v.v) || std::holds_alternative(v.v) || + std::holds_alternative(v.v) || std::holds_alternative(v.v) || + std::holds_alternative(v.v); +} + /// True when every arg in [first, last) is integral (recurses into lists). bool all_int(std::span args) { for (auto &a : args) { @@ -464,6 +473,7 @@ value_t builtin_cdr(std::span args) { value_t builtin_cons(std::span args) { expect_exact(args, 2, "cons"); auto &lst = as_list(args[1], "cons"); + eval_alloc_note(alloc_bytes(lst.size() + 1, sizeof(value_t))); // a copy: heap accounting std::vector result; result.reserve(1 + lst.size()); result.push_back(args[0]); @@ -487,6 +497,8 @@ value_t builtin_set_nth(std::span args) { if (idx < 0 || static_cast(idx) >= lst.size()) throw std::runtime_error("set-nth: index out of range"); lst[static_cast(idx)] = args[2]; + if (can_close_cycle(args[2])) + track_mutated(std::get(args[0].v)); return args[2]; } @@ -505,6 +517,8 @@ value_t builtin_append(std::span args) { expect_exact(args, 2, "append"); auto &lst = as_list_mut(args[0], "append"); lst.push_back(args[1]); + if (can_close_cycle(args[1])) + track_mutated(std::get(args[0].v)); return args[0]; } @@ -524,6 +538,7 @@ value_t builtin_slice(std::span args) { end = static_cast(lst.size()); if (start >= end) return make_list(); + eval_alloc_note(alloc_bytes(static_cast(end - start), sizeof(value_t))); // a copy return make_list(std::vector(lst.begin() + start, lst.begin() + end)); } @@ -572,6 +587,8 @@ value_t builtin_set_attr(std::span args) { expect_exact(args, 3, "set-attr"); auto &d = as_dict_mut(args[0], "set-attr"); const std::string &key = as_string(args[1], "set-attr"); + if (can_close_cycle(args[2])) + track_mutated(std::get(args[0].v)); for (auto &[k, v] : d) { if (k == key) { v = args[2]; @@ -694,6 +711,7 @@ value_t builtin_generator(std::span args) { } gen->state = gen->evaluator.create_state(body_expr, env); gen->state.stats.start(); + track_captured_env(env); // bound back where the closure can reach, it is a cycle return value_t{std::move(gen)}; } diff --git a/src/cvc/state/state_exec/closure_natives.h b/src/cvc/state/state_exec/closure_natives.h new file mode 100644 index 000000000..a92955f5b --- /dev/null +++ b/src/cvc/state/state_exec/closure_natives.h @@ -0,0 +1,39 @@ +// Private: the native functions defclass makes, as named types rather than lambdas, so the cycle +// collector (heap.cpp) can see what they hold through std::function::target(). A lambda's +// captures are invisible to it, and a class's methods capture the environment the class is bound +// in, which closes a cycle (env -> class dict -> method -> closure -> env). +#ifndef CVC_STATE_EXEC_CLOSURE_NATIVES_H +#define CVC_STATE_EXEC_CLOSURE_NATIVES_H + +#include +#include +#include + +namespace cvc::state_exec::detail { + +/// A defclass method: runs `method` (self first, then the call's arguments) on a fresh evaluator +/// over `global_env`. +struct class_method { + closure_ptr method; + environment_ptr global_env; + value_t operator()(std::span args) const; +}; + +/// A defclass constructor: a new instance dict whose __class__ is `cls_dict`, passed to the +/// class's init method if it has one. +struct class_constructor { + dict_ptr cls_dict; + value_t operator()(std::span init_args) const; +}; + +/// The init or finalize a class gets when it defines none: the superclass's `name` method, if +/// `super_cls` has one; otherwise init returns self (the first argument) and finalize nil. +struct class_inherited_method { + dict_ptr super_cls; // may be null: no superclass + std::string name; // "init" or "finalize" + value_t operator()(std::span args) const; +}; + +} // namespace cvc::state_exec::detail + +#endif // CVC_STATE_EXEC_CLOSURE_NATIVES_H diff --git a/src/cvc/state/state_exec/heap.cpp b/src/cvc/state/state_exec/heap.cpp new file mode 100644 index 000000000..b2b599ca7 --- /dev/null +++ b/src/cvc/state/state_exec/heap.cpp @@ -0,0 +1,643 @@ +#include "closure_natives.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cvc::state_exec { + +namespace { + +// --------------------------------------------------------------------------- +// The object graph +// --------------------------------------------------------------------------- + +// The reference-counted objects values point at. Strings, numbers and symbols live inside their +// value_t and are sized with whatever holds it. A native function is opaque, except the defclass +// natives (closure_natives.h), whose captures are edges. A frozen scope (the shared stdlib) is +// left out: it is never garbage and belongs to no process. +enum class node_kind : std::uint8_t { env, closure, list, dict, generator, data }; + +using dict_vector = std::vector>; + +template void env_edge(const environment_ptr &e, F &f) { + if (e && !e->frozen) + f(node_kind::env, e); +} + +// Call f(kind, shared_ptr) for each object `v` holds directly. +template void value_edges(const value_t &v, F &f) { + std::visit( + [&](const auto &x) { + using T = std::decay_t; + if constexpr (std::is_same_v) { + if (x) + f(node_kind::list, x); + } else if constexpr (std::is_same_v) { + if (x) + f(node_kind::dict, x); + } else if constexpr (std::is_same_v) { + if (x) + f(node_kind::closure, x); + } else if constexpr (std::is_same_v) { + if (x) + f(node_kind::generator, x); + } else if constexpr (std::is_same_v) { + if (x) + f(node_kind::data, x); + } else if constexpr (std::is_same_v) { + if (const auto *m = x.template target()) { + if (m->method) + f(node_kind::closure, m->method); + env_edge(m->global_env, f); + } else if (const auto *c = x.template target()) { + if (c->cls_dict) + f(node_kind::dict, c->cls_dict); + } else if (const auto *i = x.template target()) { + if (i->super_cls) + f(node_kind::dict, i->super_cls); + } + } + }, + v.v); +} + +template void frame_edges(const eval_frame &fr, F &f) { + value_edges(fr.expr, f); + env_edge(fr.env, f); + for (const value_t &v : fr.results) + value_edges(v, f); + for (const value_t &v : fr.extra_vals) + value_edges(v, f); + for (const std::vector &l : fr.extra_lists) + for (const value_t &v : l) + value_edges(v, f); + env_edge(fr.extra_env, f); +} + +template void state_edges(const evaluator_state &st, F &f) { + for (const eval_frame &fr : st.stack) + frame_edges(fr, f); + value_edges(st.result, f); + value_edges(st.root_expr, f); + env_edge(st.global_env, f); + for (const auto &[name, macro] : st.user_macros) + value_edges(macro.second, f); +} + +// Call f(kind, shared_ptr) for each object the node `p` (of kind `k`) holds directly. +template void node_edges(node_kind k, const void *p, F &f) { + switch (k) { + case node_kind::env: { + const auto &e = *static_cast(p); + for (const auto &[name, v] : e.bindings) + value_edges(v, f); + env_edge(e.outer, f); + break; + } + case node_kind::closure: { + const auto &c = *static_cast(p); + env_edge(c.env_snapshot, f); + for (const value_t &v : c.body) + value_edges(v, f); + break; + } + case node_kind::list: + for (const value_t &v : *static_cast *>(p)) + value_edges(v, f); + break; + case node_kind::dict: + for (const auto &[key, v] : *static_cast(p)) + value_edges(v, f); + break; + case node_kind::generator: { + const auto &g = *static_cast(p); + env_edge(g.evaluator.global_env(), f); + state_edges(g.state, f); + break; + } + case node_kind::data: + break; + } +} + +// std::queue keeps its container protected; this reads it in place. +template const std::deque &queue_items(const std::queue &q) { + struct access : std::queue { + static const std::deque &items(const std::queue &q) { return q.*&access::c; } + }; + return access::items(q); +} + +// Every reference a process holds directly: its roots. +template void process_edges(const process &p, F &f) { + state_edges(p.state, f); + for (const value_t &v : queue_items(p.inbox)) + value_edges(v, f); + value_edges(p.exit_code, f); + for (const auto &[name, v] : p.signal_handlers) + value_edges(v, f); + for (const eval_frame &fr : p.saved_stack) + frame_edges(fr, f); + value_edges(p.saved_result, f); + for (const auto &[id, w] : p.watch_handlers) + value_edges(w.handler, f); +} + +// --------------------------------------------------------------------------- +// Sizes: an estimate of the heap bytes each object owns (malloc's own overhead not counted) +// --------------------------------------------------------------------------- + +const std::size_t kSso = std::string().capacity(); // a string's inline capacity +constexpr std::size_t kControlBlock = 4 * sizeof(void *); // a shared_ptr's counts and vtable + +std::size_t string_heap(const std::string &s) { return s.capacity() > kSso ? s.capacity() + 1 : 0; } + +std::size_t inline_heap(const value_t &v) { + if (const auto *s = std::get_if(&v.v)) + return string_heap(*s); + if (const auto *b = std::get_if(&v.v)) + return string_heap(b->data); + if (const auto *y = std::get_if(&v.v)) + return string_heap(y->name); + return 0; +} + +std::size_t values_bytes(const std::vector &vs) { + std::size_t n = vs.capacity() * sizeof(value_t); + for (const value_t &v : vs) + n += inline_heap(v); + return n; +} + +std::size_t frame_bytes(const eval_frame &fr) { + std::size_t n = inline_heap(fr.expr) + values_bytes(fr.results) + values_bytes(fr.extra_vals); + n += fr.extra_strs.capacity() * sizeof(std::string); + for (const std::string &s : fr.extra_strs) + n += string_heap(s); + n += fr.extra_lists.capacity() * sizeof(std::vector); + for (const std::vector &l : fr.extra_lists) + n += values_bytes(l); + return n; +} + +std::size_t state_bytes(const evaluator_state &st) { + std::size_t n = st.stack.capacity() * sizeof(eval_frame); + for (const eval_frame &fr : st.stack) + n += frame_bytes(fr); + return n + inline_heap(st.result); +} + +std::size_t node_bytes(node_kind k, const void *p) { + switch (k) { + case node_kind::env: { + const auto &e = *static_cast(p); + std::size_t n = + sizeof(environment) + kControlBlock + e.bindings.bucket_count() * sizeof(void *); + n += e.bindings.size() * + (sizeof(std::pair) + 2 * sizeof(void *)); // the map's nodes + for (const auto &[name, v] : e.bindings) + n += string_heap(name) + inline_heap(v); + return n; + } + case node_kind::closure: { + const auto &c = *static_cast(p); + std::size_t n = sizeof(closure) + kControlBlock + c.params.capacity() * sizeof(symbol); + for (const symbol &s : c.params) + n += string_heap(s.name); + return n + values_bytes(c.body); + } + case node_kind::list: + return sizeof(std::vector) + kControlBlock + + values_bytes(*static_cast *>(p)); + case node_kind::dict: { + const auto &d = *static_cast(p); + std::size_t n = + sizeof(dict_vector) + kControlBlock + d.capacity() * sizeof(dict_vector::value_type); + for (const auto &[key, v] : d) + n += string_heap(key) + inline_heap(v); + return n; + } + case node_kind::generator: + return sizeof(generator) + kControlBlock + + state_bytes(static_cast(p)->state); + case node_kind::data: + return sizeof(data_object) + kControlBlock + + string_heap(static_cast(p)->type_name); + } + return 0; +} + +// What a process holds in its own structures (frames, inbox, handlers), not in shared objects. +std::size_t process_own_bytes(const process &p) { + std::size_t n = state_bytes(p.state) + inline_heap(p.exit_code) + inline_heap(p.saved_result); + const std::deque &inbox = queue_items(p.inbox); + n += inbox.size() * sizeof(value_t); + for (const value_t &v : inbox) + n += inline_heap(v); + n += p.saved_stack.capacity() * sizeof(eval_frame); + for (const eval_frame &fr : p.saved_stack) + n += frame_bytes(fr); + return n; +} + +// --------------------------------------------------------------------------- +// Measuring a share +// --------------------------------------------------------------------------- + +// A process's share of what it reaches: each node's size x the fraction of its owners that are the +// process's. A root reference (a frame, a binding the process holds) gives 1 / owners; a node +// passes its fraction on to the nodes it holds, each edge giving fraction / (child's owners). The +// fractions are added in topological order (Kahn) once the discovery pass has counted each node's +// incoming references from this process; nodes left waiting (a cycle, or a graph that changed +// between slices) are taken as they are. The job holds every node it found, and subtracts that +// one reference when it reads a node's owner count. +class heap_share_job final : public data_job { +public: + explicit heap_share_job(const process &p) : own_(static_cast(process_own_bytes(p))) { + // The roots, whole: the references the process holds directly. + auto root = [this](node_kind k, const auto &sp) { add(k, sp, /*from_root=*/true); }; + process_edges(p, root); + } + + bool step(std::size_t budget) override { + std::size_t visited = 0; + const auto more = [&] { return budget == 0 || visited < budget; }; + while (phase_ == phase::discover && more()) { + if (cursor_ == graph_.size()) { + start_propagating(); + break; + } + const std::size_t i = cursor_++; + auto edge = [this](node_kind k, const auto &sp) { add(k, sp, /*from_root=*/false); }; + node_edges(graph_[i].kind, graph_[i].hold.get(), edge); + ++visited; + ++nodes_; // data_job: nodes visited + } + while (phase_ == phase::propagate && more()) { + if (ready_.empty()) { + while (cursor_ < graph_.size() && graph_[cursor_].done) + ++cursor_; + if (cursor_ == graph_.size()) { + phase_ = phase::done; + break; + } + ready_.push_back(cursor_); // waiting on references that never come: take it as it is + } + const std::size_t i = ready_.back(); + ready_.pop_back(); + if (graph_[i].done) + continue; + graph_[i].done = true; + const double frac = std::min(1.0, graph_[i].frac); + share_ += frac * static_cast(node_bytes(graph_[i].kind, graph_[i].hold.get())); + auto edge = [&](node_kind, const auto &sp) { + auto it = index_.find(sp.get()); + if (it == index_.end()) + return; // appeared since discovery + node &c = graph_[it->second]; + if (c.done) + return; + c.frac += frac / c.owners; + if (c.pending > 0 && --c.pending == 0) + ready_.push_back(it->second); + }; + node_edges(graph_[i].kind, graph_[i].hold.get(), edge); + ++visited; + ++nodes_; // data_job: nodes visited + } + return phase_ == phase::done; + } + + value_t result() override { return value_t{static_cast(std::llround(share_ + own_))}; } + +private: + struct node { + node_kind kind; + std::shared_ptr hold; + double owners; // the node's use count when found, less this job's reference + std::size_t in = 0; // references to it from the process's roots and reachable nodes + std::size_t roots = 0; // of which from the roots + std::size_t pending = 0; // references whose fraction has not arrived yet + double frac = 0.0; + bool done = false; + }; + enum class phase { discover, propagate, done }; + + template void add(node_kind k, const std::shared_ptr &sp, bool from_root) { + auto [it, inserted] = index_.try_emplace(sp.get(), graph_.size()); + if (inserted) { + const long owners = sp.use_count(); // read before this job holds it + graph_.push_back( + node{k, std::shared_ptr(sp), static_cast(std::max(1, owners))}); + } + node &n = graph_[it->second]; + ++n.in; + if (from_root) + ++n.roots; + } + + void start_propagating() { + for (std::size_t i = 0; i < graph_.size(); ++i) { + node &n = graph_[i]; + n.frac = static_cast(n.roots) / n.owners; + n.pending = n.in - n.roots; + if (n.pending == 0) + ready_.push_back(i); + } + std::reverse(ready_.begin(), ready_.end()); // roots first, in discovery order + cursor_ = 0; + phase_ = phase::propagate; + } + + std::vector graph_; + std::unordered_map index_; + std::vector ready_; + phase phase_ = phase::discover; + std::size_t cursor_ = 0; + double share_ = 0.0; + double own_; +}; + +// --------------------------------------------------------------------------- +// The cycle registry (per thread) +// --------------------------------------------------------------------------- + +struct tracked_object { + node_kind kind; + std::weak_ptr ptr; + const void *raw; +}; + +struct cycle_registry { + std::vector objects; + std::unordered_map index; // object -> its entry in `objects` + std::size_t since_last = 0; // newly tracked since the last collection + std::size_t survivors = 0; // tracked objects alive after the last collection + std::size_t last_examined = 0; // objects the last collection walked + bool exit_pending = false; // a process ended since the last collection + std::chrono::steady_clock::time_point last_collection{}; + + void reindex() { + index.clear(); + for (std::size_t i = 0; i < objects.size(); ++i) + index.emplace(objects[i].raw, i); + } +}; + +cycle_registry ®istry() { + thread_local cycle_registry r; + return r; +} + +constexpr std::size_t kMinCollectionGrowth = 256; + +template void track(node_kind k, const std::shared_ptr &p) { + if (!p) + return; + cycle_registry &r = registry(); + if (auto it = r.index.find(p.get()); it != r.index.end()) { + tracked_object &t = r.objects[it->second]; + if (!t.ptr.owner_before(p) && !p.owner_before(t.ptr)) + return; // already tracked (a closure made in a loop over one scope, a list appended to) + t = tracked_object{k, std::weak_ptr(p), p.get()}; // a dead object's address, reused + ++r.since_last; + return; + } + if (r.objects.size() >= 2 * (r.survivors + r.since_last) + 1024) { + // Most entries died young: drop them before their (weak) control blocks pile up. + std::erase_if(r.objects, [](const tracked_object &t) { return t.ptr.expired(); }); + r.reindex(); + } + r.index.emplace(p.get(), r.objects.size()); + r.objects.push_back(tracked_object{k, std::weak_ptr(p), p.get()}); + ++r.since_last; +} + +void clear_garbage(node_kind k, void *p) { + switch (k) { + case node_kind::env: { + auto &e = *static_cast(p); + e.bindings.clear(); + e.outer.reset(); + break; + } + case node_kind::closure: { + auto &c = *static_cast(p); + c.env_snapshot.reset(); + c.body.clear(); + break; + } + case node_kind::list: + static_cast *>(p)->clear(); + break; + case node_kind::dict: + static_cast(p)->clear(); + break; + case node_kind::generator: { + auto &g = *static_cast(p); + g.state = evaluator_state{}; + g.native_step = nullptr; + g.exhausted = true; + break; + } + case node_kind::data: + break; + } +} + +// A size_t setting from the tree; nullopt when missing, empty or not a plain decimal number. +std::optional read_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 out = 0; + const auto [end, ec] = std::from_chars(v.data(), v.data() + v.size(), out); + if (v.empty() || ec != std::errc() || end != v.data() + v.size()) + return std::nullopt; + return out; +} + +} // namespace + +// --------------------------------------------------------------------------- +// Public interface +// --------------------------------------------------------------------------- + +heap_options heap_options_from_state(cvc::state &root, const std::string &scheduler_id, + const heap_options &base) { + heap_options opts = base; + const std::string suffix = ".limits.heap_check_steps"; + if (auto v = read_setting(root, std::string(scheduler::defaults_prefix) + suffix)) + opts.check_steps = *v; + if (!scheduler_id.empty()) + if (auto v = + read_setting(root, std::string(scheduler::sched_prefix) + "." + scheduler_id + suffix)) + opts.check_steps = *v; + return opts; +} + +std::unique_ptr make_heap_share_job(const process &proc) { + return std::make_unique(proc); +} + +std::size_t measure_heap_share(const process &proc) { + heap_share_job job(proc); + while (!job.step(0)) { + } + return static_cast(std::get(job.result().v)); +} + +std::size_t heap_remeasure_bytes(std::size_t last_measured) { + return std::max(std::size_t{64} << 10, last_measured / 2); +} + +void heap_after_step(process &proc, std::size_t allocated, uint64_t state_bytes_used) { + evaluator_state &st = proc.state; + if (st.heap_measured) { + proc.heap_bytes = *st.heap_measured; + st.heap_measured.reset(); + proc.heap_measuring = false; + proc.heap_alloc_bytes = 0; + proc.heap_steps = 0; + ++proc.heap_measurements; + return; + } + if (proc.max_memory == 0) + return; + proc.heap_alloc_bytes += allocated; + ++proc.heap_steps; + if (proc.heap_measuring) { + if (!st.data_work) + proc.heap_measuring = false; // dropped with the state it ran in (a signal, a restore) + return; + } + if (st.done || st.reading || st.data_work || st.stack.empty()) + return; + if (!proc.heap_measure_due(state_bytes_used)) + return; + st.data_work = std::make_shared( + pending_data_work{make_heap_share_job(proc), {}, pending_data_work::kind::heap_measure}); + proc.heap_measuring = true; +} + +void track_captured_env(const environment_ptr &env) { + if (env && !env->frozen) + track(node_kind::env, env); +} + +void track_mutated(const list_ptr &list) { track(node_kind::list, list); } +void track_mutated(const dict_ptr &dict) { track(node_kind::dict, dict); } + +cycle_stats collect_cycles() { + cycle_stats stats; + if (detail::data_step_scope::current()) + return stats; // inside an evaluator step: native code may hold raw pointers into the heap + cycle_registry &r = registry(); + r.since_last = 0; + r.exit_pending = false; + r.last_collection = std::chrono::steady_clock::now(); + + struct gc_node { + node_kind kind; + std::shared_ptr hold; + long owners; // use count when found, less this collection's reference + long internal = 0; // references to it from the nodes found + bool live = false; + }; + std::vector nodes; + std::unordered_map index; + + // The candidates: tracked objects still alive, once each. + std::vector alive; + alive.reserve(r.objects.size()); + for (tracked_object &t : r.objects) { + std::shared_ptr p = t.ptr.lock(); + if (!p || !index.try_emplace(p.get(), nodes.size()).second) + continue; + const long owners = p.use_count() - 1; // less `p` + nodes.push_back(gc_node{t.kind, std::move(p), owners}); + alive.push_back(std::move(t)); + } + r.objects = std::move(alive); + stats.candidates = nodes.size(); + + // Everything they reach, and how many references each gets from the others. + auto found = [&](node_kind k, const auto &sp) { + auto [it, inserted] = index.try_emplace(sp.get(), nodes.size()); + if (inserted) { + const long owners = sp.use_count(); // read before this collection holds it + nodes.push_back(gc_node{k, std::shared_ptr(sp), owners}); + } + ++nodes[it->second].internal; + }; + for (std::size_t i = 0; i < nodes.size(); ++i) { + void *p = nodes[i].hold.get(); + node_edges(nodes[i].kind, p, found); + } + stats.examined = nodes.size(); + + // Held from outside (more owners than references among the nodes): live, with all it reaches. + std::vector stack; + for (std::size_t i = 0; i < nodes.size(); ++i) + if (nodes[i].owners > nodes[i].internal) { + nodes[i].live = true; + stack.push_back(i); + } + auto reach = [&](node_kind, const auto &sp) { + auto it = index.find(sp.get()); + if (it != index.end() && !nodes[it->second].live) { + nodes[it->second].live = true; + stack.push_back(it->second); + } + }; + while (!stack.empty()) { + const std::size_t i = stack.back(); + stack.pop_back(); + node_edges(nodes[i].kind, nodes[i].hold.get(), reach); + } + + // The rest is garbage: clear it (its parts stay held here, so nothing frees mid-loop), then + // let go, and it frees. + for (gc_node &n : nodes) + if (!n.live) { + clear_garbage(n.kind, n.hold.get()); + ++stats.freed; + } + nodes.clear(); + + std::erase_if(r.objects, [](const tracked_object &t) { return t.ptr.expired(); }); + r.reindex(); + r.survivors = r.objects.size(); + r.last_examined = stats.examined; + return stats; +} + +bool cycle_collection_due() { + const cycle_registry &r = registry(); + // As many new objects as survived, and at least an eighth of what the last collection walked, + // so a collection's walk costs a bounded amount per object tracked since the one before. + if (r.since_last >= std::max({kMinCollectionGrowth, r.survivors, r.last_examined / 8})) + return true; + return r.exit_pending && r.since_last > 0 && + std::chrono::steady_clock::now() - r.last_collection >= std::chrono::seconds(1); +} + +void note_process_exit() { registry().exit_pending = true; } + +std::size_t tracked_objects() { return registry().objects.size(); } + +} // namespace cvc::state_exec diff --git a/src/cvc/state/state_exec/intrinsics.cpp b/src/cvc/state/state_exec/intrinsics.cpp index de35c0652..88c37bd59 100644 --- a/src/cvc/state/state_exec/intrinsics.cpp +++ b/src/cvc/state/state_exec/intrinsics.cpp @@ -448,6 +448,7 @@ value_t intrinsic_inspect(intrinsics_context *ctx, std::span args 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("heap_memory", value_t(static_cast(info->heap_memory))); entries.emplace_back("peak_memory", value_t(static_cast(info->peak_memory))); entries.emplace_back("max_memory", value_t(static_cast(info->max_memory))); entries.emplace_back("max_time", value_t(info->max_time)); diff --git a/src/cvc/state/state_exec/process.cpp b/src/cvc/state/state_exec/process.cpp index afa45bfe4..aa69d66c9 100644 --- a/src/cvc/state/state_exec/process.cpp +++ b/src/cvc/state/state_exec/process.cpp @@ -1,5 +1,6 @@ #include #include +#include #include namespace cvc::state_exec { @@ -55,6 +56,29 @@ std::optional process::time_left() const { return std::max(0.0, left); } +void process::release_state() { + evaluation_stats stats = std::move(state.stats); + state = evaluator_state{}; + state.stats = std::move(stats); + state.done = true; + saved_stack.clear(); + saved_result = value_t{}; + signal_handlers.clear(); + watch_handlers.clear(); + inbox = {}; + heap_measuring = false; +} + +bool process::heap_measure_due(uint64_t state_bytes) const { + if (max_memory == 0 || heap_measuring) + return false; + if (heap_alloc_bytes >= heap_remeasure_bytes(static_cast(heap_bytes))) + return true; + if (state_bytes + heap_bytes + heap_alloc_bytes >= max_memory) + return heap_alloc_bytes > 0; // what natives allocated since may have crossed the budget + return heap_check_steps > 0 && heap_steps >= heap_check_steps; +} + std::optional process::alloc_limit(std::size_t used_bytes) const { std::optional limit; if (max_alloc_bytes > 0) diff --git a/src/cvc/state/state_exec/scheduler.cpp b/src/cvc/state/state_exec/scheduler.cpp index 019f3be5d..449bb7a23 100644 --- a/src/cvc/state/state_exec/scheduler.cpp +++ b/src/cvc/state/state_exec/scheduler.cpp @@ -58,8 +58,9 @@ void scheduler::load_settings() { // --- 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) --- + // --- limits.max_alloc_bytes and limits.heap_check_steps (same resolution order) --- max_alloc_bytes_ = max_alloc_bytes_from_state(*watch_root_, id_); + heap_options_ = heap_options_from_state(*watch_root_, id_); // Publish effective values to per-scheduler subtree if (!id_.empty()) { @@ -69,6 +70,8 @@ void scheduler::load_settings() { (*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 + ".limits.heap_check_steps") + .value(std::to_string(heap_options_.check_steps)); (*watch_root_)(base + ".policy") .value(policy_ == scheduling_policy::round_robin ? "round_robin" : policy_ == scheduling_policy::priority ? "priority" @@ -110,6 +113,7 @@ int scheduler::execute(const std::string &script, const execute_options &opts) { proc.max_messages = opts.max_messages; proc.max_message_bytes = opts.max_message_bytes; proc.max_alloc_bytes = max_alloc_bytes_; + proc.heap_check_steps = heap_options_.check_steps; proc.signal_handlers = opts.signal_handlers; proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); @@ -137,6 +141,7 @@ int scheduler::execute(const value_t &expr, const execute_options &opts) { proc.max_messages = opts.max_messages; proc.max_message_bytes = opts.max_message_bytes; proc.max_alloc_bytes = max_alloc_bytes_; + proc.heap_check_steps = heap_options_.check_steps; proc.signal_handlers = opts.signal_handlers; proc.on_complete = opts.on_complete; proc.create_time = std::chrono::steady_clock::now(); @@ -205,6 +210,11 @@ process_ptr scheduler::select_process() { // --------------------------------------------------------------------------- void scheduler::execute_process_step(process &proc) { + // Between steps: free the reference cycles programs leaked, once enough piled up (heap.h). + // Checked every 64 steps: the check reads thread-local state. + if ((++cycle_check_ & 63) == 0 && cycle_collection_due()) + collect_cycles(); + // Track which process is currently executing so intrinsics can query it. current_pid_ = proc.pid; current_proc_ = processes_[proc.pid]; @@ -217,6 +227,14 @@ void scheduler::execute_process_step(process &proc) { pp.reset(); } } reset_guard{current_pid_, current_proc_}; + // Once this step is over, a process it finished lets go of what it evaluated with. + struct release_guard { + process &p; + ~release_guard() { + if (p.status == process_status::terminated || p.status == process_status::killed) + p.release_state(); + } + } release_dead{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. One @@ -249,13 +267,13 @@ void scheduler::execute_process_step(process &proc) { // same guarantee the read-lane's run() already has. max_time <= 0 means unlimited (nullopt). eval_deadline_guard step_deadline(proc.time_left()); // Bound what one native call in this step may allocate: limits.max_alloc_bytes, tightened to - // what is left of max_memory (which, between steps, only sees state-tree writes). + // what is left of max_memory (its state-tree writes plus its measured heap share). // With no budget and the default cap, an armed guard would change nothing (the default applies // unarmed), so it is armed only when it does: it costs thread-local writes every step. std::optional step_alloc; if (proc.max_memory > 0 || proc.max_alloc_bytes != kDefaultMaxAllocBytes) - step_alloc.emplace( - proc.alloc_limit(proc.max_memory > 0 ? mem_tracker_.current_bytes(proc.pid) : 0)); + step_alloc.emplace(proc.alloc_limit( + proc.max_memory > 0 ? mem_tracker_.current_bytes(proc.pid) + proc.heap_bytes : 0)); const bool charge_cpu = proc.charges_cpu_time(); const double cpu0 = charge_cpu ? thread_cpu_seconds() : 0.0; bool done = evaluator_.step(proc.state); @@ -267,6 +285,10 @@ void scheduler::execute_process_step(process &proc) { proc.state.stats.add_step_time(step_seconds); if (charge_cpu) proc.charge_cpu_time(thread_cpu_seconds() - cpu0); + // Heap accounting: take a finished measurement, start the next when due (heap.h). + if (proc.max_memory > 0 || proc.state.heap_measured) + heap_after_step(proc, step_alloc ? step_alloc->allocated() : 0, + mem_tracker_.current_bytes(proc.pid)); // 5. Check if signal/watch handler just finished if ((proc.in_signal_handler || proc.in_watch_handler) && proc.state.done) { @@ -537,8 +559,9 @@ void scheduler::check_limits(process &proc) { kill_process(proc, "time_limit_exceeded"); return; } - // max_memory - if (proc.max_memory > 0 && mem_tracker_.current_bytes(proc.pid) >= proc.max_memory) { + // max_memory: state-tree writes plus the measured heap share (heap.h) + if (proc.max_memory > 0 && + mem_tracker_.current_bytes(proc.pid) + proc.heap_bytes >= proc.max_memory) { kill_process(proc, "memory_limit_exceeded"); return; } @@ -565,12 +588,21 @@ void scheduler::terminate_process(process &proc, value_t result) { if (proc.on_complete) { proc.on_complete(proc.exit_code); } + release_if_idle(proc); } void scheduler::kill_process(process &proc, const std::string &reason) { proc.status = process_status::killed; proc.exit_error = reason; proc.state.stats.mark_complete(); + release_if_idle(proc); +} + +void scheduler::release_if_idle(process &proc) { + note_process_exit(); // what it leaked is garbage now (heap.h) + // A process stepping right now (one that killed itself) is released once its step is over. + if (proc.pid != current_pid_) + proc.release_state(); } bool scheduler::pause(int pid) { @@ -719,6 +751,7 @@ int scheduler::fork(int pid) { child->max_time = parent.max_time; child->max_memory = parent.max_memory; child->max_alloc_bytes = parent.max_alloc_bytes; + child->heap_check_steps = parent.heap_check_steps; child->max_messages = parent.max_messages; child->max_message_bytes = parent.max_message_bytes; child->signal_handlers = parent.signal_handlers; @@ -819,6 +852,8 @@ process_info scheduler::make_info(const process &proc) const { info.elapsed_time = proc.elapsed_time(); info.current_memory = mem_tracker_.current_bytes(proc.pid); info.peak_memory = mem_tracker_.peak_bytes(proc.pid); + info.heap_memory = proc.heap_bytes; + info.heap_measurements = proc.heap_measurements; info.max_memory = proc.max_memory; info.max_time = proc.max_time; info.message_count = proc.message_count; diff --git a/src/cvc/state/state_exec/stackless_evaluator.cpp b/src/cvc/state/state_exec/stackless_evaluator.cpp index 7858da2b8..d04a5cd37 100644 --- a/src/cvc/state/state_exec/stackless_evaluator.cpp +++ b/src/cvc/state/state_exec/stackless_evaluator.cpp @@ -1,6 +1,9 @@ +#include "closure_natives.h" + #include #include #include +#include #include #include #include @@ -117,6 +120,12 @@ bool stackless_evaluator::step_data(evaluator_state &state) { return false; state.data_work.reset(); value_t result = work->job->result(); + if (work->what == pending_data_work::kind::heap_measure) { + // Record the measurement for the scheduler; evaluation resumes where it was. + if (const auto *bytes = std::get_if(&result.v)) + state.heap_measured = static_cast(*bytes); + return state.done; + } if (work->what == pending_data_work::kind::macro_expansion) { begin_expansion(state, std::move(result)); return false; @@ -461,6 +470,7 @@ void stackless_evaluator::step_init(evaluator_state &state, eval_frame &frame) { cls->variadic = variadic; cls->body = std::vector(args.begin() + 1, args.end()); cls->env_snapshot = env; + track_captured_env(env); // bound back into env (or anything it holds), it is a cycle pop_frame(state, value_t{std::move(cls)}); return; } @@ -547,6 +557,7 @@ void stackless_evaluator::step_init(evaluator_state &state, eval_frame &frame) { cls->variadic = variadic; cls->body = std::vector(args.begin() + 2, args.end()); cls->env_snapshot = env; + track_captured_env(env); // bound in the env it captured: a cycle env->set(name_sym->name, value_t{cls}); pop_frame(state, value_t{std::move(cls)}); return; @@ -1088,44 +1099,16 @@ void stackless_evaluator::handle_defclass(evaluator_state &state, eval_frame &fr return false; }; - // Default init - if (!has_method("init")) { - auto sc = super_cls_dict; - native_fn default_init = [sc](std::span a) -> value_t { - if (sc) { - for (auto &[k, v] : *sc) { - if (k == "init") { - if (auto *fn = std::get_if(&v.v)) - return (*fn)(a); - break; - } - } - } - return a.empty() ? nil_value : a[0]; - }; - cls_dict->emplace_back("init", value_t{std::move(default_init)}); - } - - // Default finalize - if (!has_method("finalize")) { - auto sc = super_cls_dict; - native_fn default_finalize = [sc](std::span a) -> value_t { - if (sc) { - for (auto &[k, v] : *sc) { - if (k == "finalize") { - if (auto *fn = std::get_if(&v.v)) - return (*fn)(a); - break; - } - } - } - return nil_value; - }; - cls_dict->emplace_back("finalize", value_t{std::move(default_finalize)}); - } + // Default init and finalize: the superclass's, else init returns self and finalize nil. The + // natives are named types (closure_natives.h) so the cycle collector sees what they hold. + if (!has_method("init")) + cls_dict->emplace_back( + "init", value_t{native_fn{detail::class_inherited_method{super_cls_dict, "init"}}}); + if (!has_method("finalize")) + cls_dict->emplace_back( + "finalize", value_t{native_fn{detail::class_inherited_method{super_cls_dict, "finalize"}}}); // Process methods - stackless_evaluator *self = this; for (size_t i = methods_start; i < args.size(); ++i) { auto *ml = std::get_if(&args[i].v); if (!ml || !*ml || (**ml).empty()) @@ -1155,55 +1138,68 @@ void stackless_evaluator::handle_defclass(evaluator_state &state, eval_frame &fr mcls->variadic = variadic; mcls->body = std::vector(mlist.begin() + 2, mlist.end()); mcls->env_snapshot = env; + track_captured_env(env); // the class is bound in env: a cycle candidate - auto genv = global_env_; - native_fn wrapper = [self, mcls, genv](std::span call_args) -> value_t { - // Create a temp state and run to completion - std::vector all_args(call_args.begin(), call_args.end()); - auto local = environment::extend(mcls->env_snapshot); - if (mcls->variadic) { - local->set(mcls->params.back().name, make_list(all_args)); - } else { - for (size_t j = 0; j < mcls->params.size(); ++j) - local->set(mcls->params[j].name, all_args[j]); - } - if (mcls->body.size() == 1) { - // Need a recursive evaluation — use a temporary evaluator - stackless_evaluator temp(genv); - return temp.evaluate(mcls->body[0], local); - } - std::vector be; - be.reserve(1 + mcls->body.size()); - be.push_back(value_t{symbol{"begin"}}); - be.insert(be.end(), mcls->body.begin(), mcls->body.end()); - stackless_evaluator temp(genv); - return temp.evaluate(make_list(std::move(be)), local); - }; - cls_dict->emplace_back(msym->name, value_t{std::move(wrapper)}); + cls_dict->emplace_back(msym->name, + value_t{native_fn{detail::class_method{std::move(mcls), global_env_}}}); } // Constructor - auto cls_dict_val = value_t{cls_dict}; - native_fn constructor = [cls_dict, cls_dict_val](std::span init_args) -> value_t { - auto inst = std::make_shared>>(); - inst->emplace_back("__class__", cls_dict_val); - value_t inst_val{inst}; - for (auto &[k, v] : *cls_dict) { - if (k == "init") { - std::vector a; - a.push_back(inst_val); - a.insert(a.end(), init_args.begin(), init_args.end()); + env->set(cls_name, value_t{native_fn{detail::class_constructor{cls_dict}}}); + env->set(cls_name + ".__cls_dict__", value_t{cls_dict}); + pop_frame(state, *env->lookup(cls_name)); +} + +value_t detail::class_method::operator()(std::span call_args) const { + // Run the method to completion on a temporary evaluator. + std::vector all_args(call_args.begin(), call_args.end()); + auto local = environment::extend(method->env_snapshot); + if (method->variadic) { + local->set(method->params.back().name, make_list(all_args)); + } else { + for (size_t j = 0; j < method->params.size(); ++j) + local->set(method->params[j].name, all_args[j]); + } + stackless_evaluator temp(global_env); + if (method->body.size() == 1) + return temp.evaluate(method->body[0], local); + std::vector be; + be.reserve(1 + method->body.size()); + be.push_back(value_t{symbol{"begin"}}); + be.insert(be.end(), method->body.begin(), method->body.end()); + return temp.evaluate(make_list(std::move(be)), local); +} + +value_t detail::class_constructor::operator()(std::span init_args) const { + auto inst = std::make_shared>>(); + inst->emplace_back("__class__", value_t{cls_dict}); + value_t inst_val{inst}; + for (auto &[k, v] : *cls_dict) { + if (k == "init") { + std::vector a; + a.push_back(inst_val); + a.insert(a.end(), init_args.begin(), init_args.end()); + if (auto *fn = std::get_if(&v.v)) + (*fn)(a); + break; + } + } + return inst_val; +} + +value_t detail::class_inherited_method::operator()(std::span args) const { + if (super_cls) { + for (auto &[k, v] : *super_cls) { + if (k == name) { if (auto *fn = std::get_if(&v.v)) - (*fn)(a); + return (*fn)(args); break; } } - return inst_val; - }; - - env->set(cls_name, value_t{std::move(constructor)}); - env->set(cls_name + ".__cls_dict__", cls_dict_val); - pop_frame(state, *env->lookup(cls_name)); + } + if (name == "init") + return args.empty() ? nil_value : args[0]; + return nil_value; } } // namespace cvc::state_exec diff --git a/src/cvc/state/state_exec/types.cpp b/src/cvc/state/state_exec/types.cpp index 1bb94300d..39717cc51 100644 --- a/src/cvc/state/state_exec/types.cpp +++ b/src/cvc/state/state_exec/types.cpp @@ -270,6 +270,12 @@ void eval_alloc_check(std::size_t bytes, const char *who) { throw std::runtime_error(std::string(who) + ": the result needs " + std::to_string(bytes) + " bytes, over the allocation limit of " + std::to_string(*limit) + " bytes (limits.max_alloc_bytes, or what is left of max_memory)"); + eval_alloc_note(bytes); +} + +void eval_alloc_note(std::size_t bytes) { + if (eval_alloc_guard *g = t_alloc_innermost) + g->allocated_ = g->allocated_ > SIZE_MAX - bytes ? SIZE_MAX : g->allocated_ + bytes; } // -- value_tag --------------------------------------------------------------- diff --git a/src/cvc/tests/CMakeLists.txt b/src/cvc/tests/CMakeLists.txt index 7e99e1f04..d312d5f62 100644 --- a/src/cvc/tests/CMakeLists.txt +++ b/src/cvc/tests/CMakeLists.txt @@ -67,6 +67,7 @@ add_executable(state_exec_stackless_evaluator_test state_exec_stackless_evaluato add_executable(state_exec_codec_test state_exec_codec_test.cpp) add_executable(state_exec_unicode_test state_exec_unicode_test.cpp) add_executable(state_exec_stdlib_test state_exec_stdlib_test.cpp) +add_executable(state_exec_heap_test state_exec_heap_test.cpp) add_executable(state_exec_grammar_test state_exec_grammar_test.cpp) add_executable(state_exec_corpus_test state_exec_corpus_test.cpp) add_executable(state_exec_pegtl_test state_exec_pegtl_test.cpp) @@ -349,6 +350,7 @@ list(APPEND TEST_TARGETS state_exec_codec_test state_exec_unicode_test state_exec_stdlib_test + state_exec_heap_test state_exec_grammar_test state_exec_corpus_test state_exec_pegtl_test @@ -1290,6 +1292,12 @@ target_link_libraries(state_exec_stdlib_test GTest::gtest GTest::gtest_main ) +target_link_libraries(state_exec_heap_test + PRIVATE + cvc + GTest::gtest + GTest::gtest_main +) target_link_libraries(state_exec_grammar_test PRIVATE cvc @@ -1687,6 +1695,7 @@ target_compile_features(state_exec_stackless_evaluator_test PRIVATE cxx_std_20) target_compile_features(state_exec_codec_test PRIVATE cxx_std_20) target_compile_features(state_exec_unicode_test PRIVATE cxx_std_20) target_compile_features(state_exec_stdlib_test PRIVATE cxx_std_20) +target_compile_features(state_exec_heap_test PRIVATE cxx_std_20) target_compile_features(state_exec_grammar_test PRIVATE cxx_std_20) target_compile_features(state_exec_corpus_test PRIVATE cxx_std_20) target_compile_features(state_exec_pegtl_test PRIVATE cxx_std_20) @@ -2109,6 +2118,7 @@ cvc_discover_tests(state_exec_stackless_evaluator_test) cvc_discover_tests(state_exec_codec_test) cvc_discover_tests(state_exec_unicode_test) cvc_discover_tests(state_exec_stdlib_test) +cvc_discover_tests(state_exec_heap_test) cvc_discover_tests(state_exec_grammar_test) cvc_discover_tests(state_exec_corpus_test) cvc_discover_tests(state_exec_pegtl_test) diff --git a/src/cvc/tests/ariadne_runtime_test.cpp b/src/cvc/tests/ariadne_runtime_test.cpp index dbca01741..67e236eea 100644 --- a/src/cvc/tests/ariadne_runtime_test.cpp +++ b/src/cvc/tests/ariadne_runtime_test.cpp @@ -22,6 +22,7 @@ #include #include // exec_scheduler().post_message end-to-end test #include // host-intrinsic seam test: register_fn +#include // collect_cycles / tracked_objects #include // intrinsics_context.root_path (pool-task test) #include // value_t #include @@ -1722,6 +1723,55 @@ TEST(AriadneStdlib, ActionMemoryBudgetIsATreeSetting) { EXPECT_EQ(click(), "10000"); } +// A finished action's process is reaped from the app scheduler, and a function it defined (bound +// in the scope it captured: a reference cycle) is freed by the cycle collector. +TEST(AriadneHeap, FinishedActionsAreReapedAndTheirCyclesFreed) { + cvc::app app; + Runtime rt(app, ""); + MockBackend mb; + rt.set_backend(&mb); + rt.set_root(group({button("Go", "(begin (defun helper (x) x) (state-set \"n\" (helper 7)))")})); + cvc::state_exec::collect_cycles(); // earlier tests' leftovers + const std::size_t before = cvc::state_exec::tracked_objects(); + for (int i = 0; i < 20; ++i) { + mb.button_click = true; + rt.render(); + rt.drain(); + } + EXPECT_EQ(cvc::state::instance(app)("n").value(), "7"); + EXPECT_EQ(app.exec_scheduler().process_count(), 0); // every finished action was reaped + cvc::state_exec::collect_cycles(); + EXPECT_LE(cvc::state_exec::tracked_objects(), before); // the 20 action scopes are gone +} + +// An action's live values count against its memory budget, not only its state-tree writes. +TEST(AriadneHeap, AnActionsLiveValuesCountAgainstItsBudget) { + cvc::app app; + Runtime rt(app, ""); + MockBackend mb; + rt.set_backend(&mb); + auto &root = cvc::state::instance(app); + rt.set_root(group({button("Keep", "(begin (set keep (list)) " + "(append keep (collections.range 1000)) " + "(append keep (collections.range 1000)) " + "(append keep (collections.range 1000)) " + "(append keep (collections.range 1000)) " + "(append keep (collections.range 1000)) " + "(append keep (collections.range 1000)) " + "(state-set \"n\" \"kept\"))")})); + const auto click = [&] { + root("n").value(std::string("-")); + mb.button_click = true; + rt.render(); + rt.drain(); + return root("n").value(); + }; + EXPECT_EQ(click(), "kept"); // well under the 4 MiB default + root("ariadne.limits.action_max_memory").value(std::string("200000")); + EXPECT_EQ(click(), "-"); // 6000 live elements are over 200 KB: killed before the write + EXPECT_FALSE(rt.take_reactive_warnings().empty()); +} + TEST(AriadneReactive, DeniedSpecialFormsAreRejectedNotRun) { cvc::app app; Runtime rt(app, ""); diff --git a/src/cvc/tests/state_exec_heap_test.cpp b/src/cvc/tests/state_exec_heap_test.cpp new file mode 100644 index 000000000..865306573 --- /dev/null +++ b/src/cvc/tests/state_exec_heap_test.cpp @@ -0,0 +1,365 @@ +/// @file state_exec_heap_test.cpp +/// @brief A process's heap: freeing leaked reference cycles, and its measured share for +/// max_memory (heap.h). + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace cvc::state_exec; + +namespace { + +// Run `script` to completion with a fresh default environment, keep weak handles to the +// environment and to the list bound to `big` (if any), and drop everything else. +struct leak_probe { + std::weak_ptr env; + std::weak_ptr> big; +}; + +leak_probe run_and_drop(const std::string &script) { + leak_probe probe; + scheduler sched; + execute_options opts; + opts.env = builtins::make_default_environment(); + probe.env = opts.env; + sched.execute(script, opts); + sched.run(); + if (const value_t *v = opts.env->lookup("big")) + if (const auto *l = std::get_if(&v->v)) + probe.big = *l; + return probe; // the scheduler, its processes and `opts` go here +} + +list_ptr int_list(std::size_t n) { + std::vector v; + v.reserve(n); + for (std::size_t i = 0; i < n; ++i) + v.emplace_back(static_cast(i)); + return std::get(make_list(std::move(v)).v); +} + +// A process whose global environment (a default one, which only the process holds) has +// `bindings`: enough for measuring. +process proc_holding(std::vector> bindings) { + auto env = builtins::make_default_environment(); + for (auto &[name, v] : bindings) + env->set(name, std::move(v)); + process p; + p.state.global_env = std::move(env); + return p; +} + +double ratio(std::size_t a, std::size_t b) { + return static_cast(a) / static_cast(b); +} + +} // namespace + +// --- reference cycles -------------------------------------------------------------------------- + +TEST(CycleCollectorTest, NoCycleNeedsNoCollector) { + const leak_probe p = run_and_drop("(set big (collections.range 1000))"); + EXPECT_TRUE(p.env.expired()); + EXPECT_TRUE(p.big.expired()); +} + +TEST(CycleCollectorTest, EveryLeakPatternIsFreed) { + // Each of these used to keep its whole environment (and the 1000-element list in it) alive + // forever once the program ended: a closure bound in the scope it captured, a container that + // holds itself, a class whose methods capture the scope it is bound in, a generator stored where + // its closure can reach it. + const char *const scripts[] = { + "(begin (set big (collections.range 1000)) (defun f (x) x))", + "(begin (set big (collections.range 1000)) (set g (lambda (x) x)))", + "(begin (set big (collections.range 1000)) (let ((h nil)) (set h (lambda () h)) 1))", + "(begin (set big (collections.range 1000)) (append big big))", + "(begin (set big (collections.range 1000)) (set d (dict)) (set-attr d \"self\" d) " + "(set-attr d \"big\" big))", + "(begin (set big (collections.range 1000)) (defclass Point (init (self x) (set-attr self " + "\"x\" x) self) (getx (self) (get-attr self \"x\"))) (set p (Point 3)))", + "(begin (set big (collections.range 1000)) (set gen (generator (lambda () (yield big)))))", + }; + for (const char *script : scripts) { + const leak_probe p = run_and_drop(script); + collect_cycles(); + EXPECT_TRUE(p.env.expired()) << script; + EXPECT_TRUE(p.big.expired()) << script; + } +} + +TEST(CycleCollectorTest, ClassesStillWork) { + // defclass's natives became named types (so the collector can see their captures); classes, + // inheritance and the default init/finalize behave as before. + scheduler sched; + const int pid = + sched.execute(std::string("(begin (defclass A (init (self v) (set-attr self \"v\" v) self) " + "(get (self) (get-attr self \"v\"))) " + "(defclass B A (twice (self) (* 2 (send self \"get\")))) " + "(defclass C) (set c (C)) " + "(list (send (A 5) \"get\") (send (B 7) \"twice\") (is-null " + "(get-attr c \"__class__\"))))"), + {}); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_EQ(to_string(*sched.get_result(pid)), "(5 14 #f)"); +} + +TEST(CycleCollectorTest, LiveDataIsNeverFreed) { + // A host-held environment (pycvc's Exec, an Ariadne action context) with a function in it. + auto env = builtins::make_default_environment(); + { + scheduler sched; + execute_options opts; + opts.env = env; + sched.execute(std::string("(begin (set base 40) (defun add2 (x) (+ x base)))"), opts); + sched.run(); + } + const cycle_stats s = collect_cycles(); + EXPECT_GE(s.candidates, 1u); + scheduler sched; + execute_options opts; + opts.env = env; + const int pid = sched.execute(std::string("(add2 2)"), opts); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_EQ(std::get(sched.get_result(pid)->v), 42); +} + +TEST(CycleCollectorTest, ParkedProcessKeepsItsFunctions) { + collect_cycles(); // earlier tests' leftovers + async_scheduler sched; + intrinsics_context ictx; + ictx.sched = &sched; + ictx.proc = make_process(); + execute_options opts; + opts.env = builtins::make_default_environment(); + register_intrinsics(opts.env, &ictx); + const int pid = sched.execute(std::string("(begin (defun twice (x) (* 2 x)) " + "(twice (get-attr (msg-recv \"in\") \"n\")))"), + opts); + sched.sync_run(); + const cycle_stats s = collect_cycles(); // the process is parked: everything it holds is live + EXPECT_EQ(s.freed, 0u); + sched.post_message("in", make_dict({{"n", value_t(int64_t(21))}})); + sched.drain_ingress(); + sched.sync_run(); + ASSERT_TRUE(sched.get_result(pid)) << sched.get_exit_error(pid).value_or(""); + EXPECT_EQ(std::get(sched.get_result(pid)->v), 42); +} + +TEST(CycleCollectorTest, DoesNothingInsideAStep) { + auto env = builtins::make_default_environment(); + cycle_stats inside{1, 1, 1}; + builtins::register_fn(env, "collect-now", [&inside](std::span) -> value_t { + inside = collect_cycles(); + return nil_value; + }); + scheduler sched; + execute_options opts; + opts.env = env; + sched.execute(std::string("(begin (defun f () 1) (collect-now))"), opts); + sched.run(); + EXPECT_EQ(inside.candidates, 0u); + EXPECT_EQ(inside.freed, 0u); +} + +TEST(CycleCollectorTest, SchedulersCollectOnTheirOwn) { + // Each of these programs leaks a let scope per iteration; the scheduler collects between steps + // once enough piled up, so the first iterations' scopes are gone by the end without any + // explicit collect_cycles(). + collect_cycles(); + std::weak_ptr first; + bool remembered = false; + scheduler sched; + auto env = builtins::make_default_environment(); + builtins::register_fn(env, "remember", + [&first, &remembered](std::span a) -> value_t { + if (!remembered) + if (const auto *c = std::get_if(&a[0].v)) { + first = (*c)->env_snapshot; // the first iteration's let scope + remembered = true; + } + return nil_value; + }); + execute_options opts; + opts.env = env; + const int pid = sched.execute( + std::string("(begin (set i 0) (while (< i 2000) (begin (let ((h nil)) (set h (lambda () h)) " + "(remember h)) (set i (+ i 1)))) i)"), + opts); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_TRUE(first.expired()); + EXPECT_LT(tracked_objects(), 2000u); +} + +// --- measuring a share ------------------------------------------------------------------------- + +TEST(HeapShareTest, CountsWhatTheProcessHolds) { + // Each process is built in a statement of its own: a braced list's elements live to the end of + // the full-expression, and would be a second owner while measuring. + const process empty = proc_holding({}); + const process holder = proc_holding({{"big", value_t{int_list(10000)}}}); + const std::size_t base = measure_heap_share(empty); + const std::size_t list = measure_heap_share(holder) - base; + EXPECT_GE(list, 10000 * sizeof(value_t)); + EXPECT_LT(list, 10000 * sizeof(value_t) + 4096); +} + +TEST(HeapShareTest, SharedValuesAreSplitBetweenHolders) { + const process empty = proc_holding({}); + const process alone_holder = proc_holding({{"v", value_t{int_list(10000)}}}); + const std::size_t base = measure_heap_share(empty); + const std::size_t alone = measure_heap_share(alone_holder) - base; + list_ptr shared = int_list(10000); + std::vector holders; + for (int i = 0; i < 4; ++i) { + process p = proc_holding({{"v", value_t{shared}}}); + holders.push_back(std::move(p)); + } + shared.reset(); // the four processes are its only owners + std::size_t total = 0; + for (const process &p : holders) { + const std::size_t share = measure_heap_share(p) - base; + EXPECT_NEAR(ratio(share, alone), 0.25, 0.02) << share << " of " << alone; + total += share; + } + EXPECT_NEAR(ratio(total, alone), 1.0, 0.05); // the shares add up to the value, once +} + +TEST(HeapShareTest, ASharedNodePassesItsShareDown) { + // Two holders of an outer list whose inner list only the outer one holds: each pays half of + // both, not half of the outer and all of the inner. + const auto nested = [] { + return std::get(make_list({value_t{int_list(5000)}, value_t(int64_t(1))}).v); + }; + const process empty = proc_holding({}); + const process whole_holder = proc_holding({{"o", value_t{nested()}}}); + list_ptr outer = nested(); + const process p1 = proc_holding({{"o", value_t{outer}}}); + const process p2 = proc_holding({{"o", value_t{outer}}}); + outer.reset(); // only the two processes hold it + const std::size_t base = measure_heap_share(empty); + const std::size_t whole = measure_heap_share(whole_holder) - base; + const std::size_t s1 = measure_heap_share(p1) - base; + const std::size_t s2 = measure_heap_share(p2) - base; + EXPECT_NEAR(ratio(s1, whole), 0.5, 0.02) << s1 << " of " << whole; + EXPECT_NEAR(ratio(s2, whole), 0.5, 0.02) << s2 << " of " << whole; +} + +TEST(HeapShareTest, OptionsComeFromTheTree) { + cvc::app app; + auto &root = cvc::state::instance(app); + EXPECT_EQ(heap_options_from_state(root, "s").check_steps, 4096u); + root("state_exec.defaults.limits.heap_check_steps").value(std::string("100")); + EXPECT_EQ(heap_options_from_state(root, "s").check_steps, 100u); + root("state_exec.schedulers.s.limits.heap_check_steps").value(std::string("0")); + EXPECT_EQ(heap_options_from_state(root, "s").check_steps, 0u); + root("state_exec.schedulers.t.limits.heap_check_steps").value(std::string("x")); + EXPECT_EQ(heap_options_from_state(root, "t").check_steps, 100u); +} + +// --- max_memory covers the heap ---------------------------------------------------------------- + +TEST(HeapBudgetTest, KeptValuesCountAgainstMaxMemory) { + // No state-tree writes at all: before heap accounting this ran to the end whatever its budget. + scheduler sched; + execute_options opts; + opts.max_memory = 256 * 1024; + const int pid = sched.execute( + std::string("(begin (set keep (list)) (set i 0) (while (< i 60) " + "(begin (append keep (collections.range 200)) (set i (+ i 1)))) (length keep))"), + opts); + // Over the budget the process is killed (memory_limit_exceeded); or, when a measurement left + // less room than the next call needs, that call fails its allocation limit (which the sync + // scheduler rethrows). Without heap accounting neither happens: nothing is written to the tree. + std::string error; + try { + sched.run(); + } catch (const std::exception &e) { + error = e.what(); + } + const auto info = sched.get_process_info(pid); + ASSERT_TRUE(info); + EXPECT_TRUE(info->status == process_status::killed || + error.find("allocation limit") != std::string::npos) + << error; + EXPECT_FALSE(sched.get_result(pid)); + EXPECT_GT(info->heap_measurements, 0u); +} + +TEST(HeapBudgetTest, TemporariesDoNotCount) { + scheduler sched; + execute_options opts; + opts.max_memory = 256 * 1024; + const int pid = sched.execute( + std::string("(begin (set i 0) (while (< i 60) (begin (set tmp (collections.range 200)) " + "(set i (+ i 1)))) i)"), + opts); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_EQ(std::get(sched.get_result(pid)->v), 60); + const auto info = sched.get_process_info(pid); + ASSERT_TRUE(info); + EXPECT_GT(info->heap_measurements, 0u); + EXPECT_LT(info->heap_memory, opts.max_memory); +} + +TEST(HeapBudgetTest, NoBudgetMeansNoMeasuring) { + scheduler sched; + const int pid = sched.execute( + std::string("(begin (set keep (list)) (set i 0) (while (< i 60) " + "(begin (append keep (collections.range 200)) (set i (+ i 1)))) (length keep))"), + {}); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_EQ(sched.get_process_info(pid)->heap_measurements, 0u); +} + +TEST(HeapBudgetTest, MeasuringRunsAsSlicedDataSteps) { + cvc::app app; + auto &root = cvc::state::instance(app); + root("state_exec.defaults.data.step_nodes").value(std::string("8")); + root("state_exec.defaults.limits.heap_check_steps").value(std::string("50")); + scheduler sched; + sched.set_watch_root(&root); + sched.load_settings(); + execute_options opts; + opts.max_memory = std::size_t{64} << 20; + const int pid = + sched.execute(std::string("(begin (set keep (list)) (set i 0) (while (< i 100) " + "(begin (append keep (list i i)) (set i (+ i 1)))) (length keep))"), + opts); + sched.run(); + ASSERT_TRUE(sched.get_result(pid)); + EXPECT_EQ(std::get(sched.get_result(pid)->v), 100); + const auto info = sched.get_process_info(pid); + ASSERT_TRUE(info); + EXPECT_GT(info->heap_measurements, 0u); + EXPECT_GT(info->data_steps, info->heap_measurements); // a measurement took several slices +} + +TEST(HeapBudgetTest, AsyncSchedulerCountsTheHeapToo) { + async_scheduler sched; + execute_options opts; + opts.max_memory = 256 * 1024; + const int pid = sched.execute( + std::string("(begin (set keep (list)) (set i 0) (while (< i 60) " + "(begin (append keep (collections.range 200)) (set i (+ i 1)))) (length keep))"), + opts); + sched.sync_run(); + const std::string error = sched.get_exit_error(pid).value_or(""); + EXPECT_TRUE(error == "memory_limit_exceeded" || + error.find("allocation limit") != std::string::npos) + << error; +}