553ca19 · knot-legacy · git.iko.soy

git.iko.soy/knot-legacy

on master help sign in

Share caller db with compiled snippets so compile results are transactional in atomic

553ca19iko1 month ago

crates/knot-jit/tests/persistence.rs+126−1

compare with master

unifiedsplit

@@ -3,7 +3,18 @@

3//! process-level effect). Builds and runs real binaries.

5mod e2e;

6−use e2e::assert_stdout;

6+use e2e::{assert_stdout, run_program};

8+/// Line-set comparison (order-independent): forked output interleaves.

9+fn assert_stdout_unordered(name: &str, src: &str, expected: &[&str]) {

10+ let (out, err, code) = run_program(name, src);

11+ assert_eq!(code, 0, "program {name} exited {code}\nstderr:\n{err}");

12+ let mut got: Vec<&str> = out.trim_end().lines().collect();

13+ let mut want: Vec<&str> = expected.to_vec();

14+ got.sort_unstable();

15+ want.sort_unstable();

16+ assert_eq!(got, want, "stdout line-set mismatch for {name}");

17+}

8#[test]

19#[test]

9fn persisted_relation_groupby() {

20fn persisted_relation_groupby() {

@@ -709,3 +720,117 @@ Rel E *emps

709 // behavior, not something this fix changes; the fix is that the *message*

720 // behavior, not something this fix changes; the fix is that the *message*

710 // is the clean overflow report, not the raw SQLite error.

721 // is the clean overflow report, not the raw SQLite error.

711}

722}

723+// ── base.compile results are transactional inside `atomic` ─────────────────

724+//

725+// A compiled snippet used to open its OWN SQLite connection to the host's db

726+// file, so its writes were invisible to — and unrolled-back by — the host's

727+// `atomic` SAVEPOINT. `base.compile` now runs the snippet on the caller's live

728+// db handle (the runtime's CURRENT_DB, set at every knot_value_call/knot_io_run

729+// boundary), so a compiled value inside an `atomic` block shares the host's

730+// transaction: it reads the atomic's uncommitted writes, and its own writes

731+// roll back if the atomic fails.

732+

733+#[test]

734+fn compile_closure_in_atomic_sees_uncommitted_writes() {

735+ // The compiled closure reads *accounts at CALL time (a `\_ ->` function),

736+ // so invoking it inside the atomic — after the atomic's own write — must

737+ // see all 3 rows, not the 1 committed before the atomic began.

738+ assert_stdout(

739+ "compile_atomic_read",

740+ r#"with {

741+Account {name Text balance (Int 1)}

742+Rel Account *accounts

743+}

744+(|

745+ *accounts = [{name "a" balance 1}]

746+ (? (base.compile "with {\nAccount {name Text balance (Int 1)}\nRel Account *accounts\n}\n(\\_ -> base.count *accounts)")

747+ Result.Ok {value countNow} (|

748+ _ <- atomic |

749+ *accounts = [{name "a" balance 1} {name "b" balance 2} {name "c" balance 3}]

750+ base.println (base.show (countNow 0))

751+ yield {}

752+ yield {})

753+ Result.Err {error e} base.println ("compile err: " ++ e))

754+ yield {})"#,

755+ // Sees the atomic's uncommitted write: 3 rows, not 1.

756+ "\"[3]\"\n{}",

757+ );

758+}

759+

760+#[test]

761+fn compile_io_write_in_atomic_rolls_back_on_panic() {

762+ // The compiled IO writes a row inside the atomic; the body then panics

763+ // (integer overflow) so the atomic rolls back. If the snippet shared the

764+ // host's connection, its write is undone and the final count stays [1].

765+ // (On the old separate-connection behavior the row would persist.)

766+ assert_stdout_unordered(

767+ "compile_atomic_rollback",

768+ r#"with {

769+Account {name Text balance (Int 1)}

770+Rel Account *accounts

771+}

772+(|

773+ *accounts = [{name "a" balance 1}]

774+ (? (base.compile "with {\nAccount {name Text balance (Int 1)}\nRel Account *accounts\n}\n(| *accounts = [{name \"sneak\" balance 999}] yield 7)")

775+ Result.Ok {value io} (|

776+ base.fork (atomic |

777+ r <- io

778+ base.println (base.show (9223372036854775807 + 1))

779+ yield {})

780+ base.sleep 300

781+ yield {})

782+ Result.Err {error e} base.println ("compile err: " ++ e))

783+ base.println ("final: " ++ (base.show (base.count *accounts)))

784+ yield {})"#,

785+ &["\"final: [1]\"", "{}"],

786+ );

787+}

788+

789+#[test]

790+fn compile_outside_atomic_still_works() {

791+ // The shared-connection change must not break the ordinary path: a

792+ // compiled pure value used outside any atomic behaves exactly as before.

793+ assert_stdout(

794+ "compile_plain",

795+ r#"(? (base.the (Result Text (Int 1)) (base.compile "6 * 7"))

796+ Result.Ok {value v} base.println (base.show v)

797+ Result.Err {error e} base.println ("err: " ++ e))"#,

798+ "\"42\"\n{}",

799+ );

800+}

801+

802+/// A compiled reader gated by `base.retry`: the snippet reads `*flag` on the

803+/// host's shared connection, so its read registers with the STM change-tracker

804+/// and the atomic is woken when the forked writer flips the flag. On the old

805+/// separate-connection behavior the snippet's read was invisible to the host's

806+/// STM, so the retry would not be driven by the snippet's read. The compiled

807+/// closure runs fresh on each retry iteration and eventually sees the flag.

808+#[test]

809+fn compile_reader_in_atomic_wakes_on_retry() {

810+ assert_stdout(

811+ "compile_atomic_retry",

812+ r#"with {

813+Flag {ready (Int 1)}

814+Rel Flag *flag

815+}

816+(|

817+ *flag = [{ready 0}]

818+ base.fork (|

819+ base.sleep 200

820+ *flag = [{ready 1}]

821+ yield {})

822+ (? (base.compile "with {\nFlag {ready (Int 1)}\nRel Flag *flag\n}\n(\\_ -> base.count *flag)")

823+ Result.Ok {value countNow} (|

824+ got <- atomic |

825+ (? (base.count (countNow 0))

826+ [1] (yield "set")

827+ _ base.retry)

828+ base.println ("compiled reader: " ++ (base.show got))

829+ yield {})

830+ Result.Err {error e} base.println ("compile err: " ++ e))

831+ yield {})"#,

832+ // Blocked in retry until the fork set the flag; the compiled read then

833+ // saw it and the atomic committed.

834+ "\"compiled reader: set\"\n{}",

835+ );

836+}

crates/knot-runtime/src/lib.rs+71−9

@@ -8804,6 +8804,11 @@ pub extern "C-unwind" fn knot_value_call(

8804) -> *mut Value {

8805 match unsafe { as_ref(func) } {

8806 Value::Function(f) => {

8807+ // Establish the callee's db as this thread's current-db for the

8808+ // duration of the call, so a `base.compile` inside it (or inside a

8809+ // deeper call) runs its snippet on this handle — sharing any

8810+ // enclosing `atomic` SAVEPOINT — then restore the caller's.

8811+ let _db_scope = RestoreDb::new(db);

8807 let fun: extern "C-unwind" fn(*mut c_void, *mut Value, *mut Value) -> *mut Value =

8812 let fun: extern "C-unwind" fn(*mut c_void, *mut Value, *mut Value) -> *mut Value =

8808 unsafe { std::mem::transmute(f.fn_ptr) };

8813 unsafe { std::mem::transmute(f.fn_ptr) };

8809 fun(db, f.env, arg)

8814 fun(db, f.env, arg)

@@ -9472,6 +9477,9 @@ pub extern "C-unwind" fn knot_io_pure(val: *mut Value) -> *mut Value {

9472/// Execute an IO thunk. If the value is not IO, return it as-is.

9477/// Execute an IO thunk. If the value is not IO, return it as-is.

9473#[unsafe(no_mangle)]

9478#[unsafe(no_mangle)]

9474pub extern "C-unwind" fn knot_io_run(db: *mut c_void, val: *mut Value) -> *mut Value {

9479pub extern "C-unwind" fn knot_io_run(db: *mut c_void, val: *mut Value) -> *mut Value {

9480+ // Current-db for the whole IO chain: any `base.compile` reached while

9481+ // running these thunks runs on this handle (sharing an enclosing atomic).

9482+ let _db_scope = RestoreDb::new(db);

9475 let mut current = val;

9483 let mut current = val;

9476 loop {

9484 loop {

9477 // Cooperative cancellation: a `race` loser exits its IO chain at the

9485 // Cooperative cancellation: a `race` loser exits its IO chain at the

@@ -11857,16 +11865,28 @@ fn compile_run(

11857 Some("no JIT linked (knot-compile-rt not registered)".into()),

11865 Some("no JIT linked (knot-compile-rt not registered)".into()),

11858 );

11866 );

11859 };

11867 };

11860− // Open a db handle for the snippet. knot_db_open records the path on first

11868+ // Run the snippet on the CALLER's live db handle when one is active

11861− // open (DB_PATH), so reuse it; a pure program (no store) gets :memory:.

11869+ // (the normal case: the host program called `base.compile` from running

11870+ // code, so CURRENT_DB holds its handle). This makes the compiled value

11871+ // share the host's connection — and any enclosing `atomic` SAVEPOINT — so

11872+ // its reads see uncommitted writes and its writes roll back with the

11873+ // transaction. When there is no caller (the compiler's const-eval thread),

11874+ // fall back to opening a connection: reuse the recorded host path so a

11862− // A snippet reading the host's persisted relations thus sees real data.

11875+ // snippet reading persisted relations sees real data, else `:memory:`.

11876+ let db = {

11877+ let cur = current_db();

11878+ if !cur.is_null() {

11879+ cur

11880+ } else {

11863− let db_path = DB_PATH.lock().unwrap_or_else(|e| e.into_inner()).clone();

11881+ let db_path = DB_PATH.lock().unwrap_or_else(|e| e.into_inner()).clone();

11864− let (pstr, plen) = if db_path.is_empty() {

11882+ let (pstr, plen) = if db_path.is_empty() {

11865− (":memory:".as_bytes().as_ptr(), ":memory:".len())

11883+ (":memory:".as_bytes().as_ptr(), ":memory:".len())

11866− } else {

11884+ } else {

11867− (db_path.as_ptr(), db_path.len())

11885+ (db_path.as_ptr(), db_path.len())

11886+ };

11887+ knot_db_open(pstr, plen)

11888+ }

11868 };

11889 };

11869− let db = knot_db_open(pstr, plen);

11870

11890

11871 // The host's expected type (source-annotation string) forwarded to the JIT,

11891 // The host's expected type (source-annotation string) forwarded to the JIT,

11872 // which runs the real-type subsumption. Empty → unpinned.

11892 // which runs the real-type subsumption. Empty → unpinned.

@@ -14272,6 +14292,9 @@ pub extern "C-unwind" fn knot_run_user_main(

14272) -> *mut c_void {

14292) -> *mut c_void {

14273 let f: extern "C-unwind" fn(*mut c_void) -> *mut c_void =

14293 let f: extern "C-unwind" fn(*mut c_void) -> *mut c_void =

14274 unsafe { std::mem::transmute(main_fn) };

14294 unsafe { std::mem::transmute(main_fn) };

14295+ // Establish the program's db as current-db so a top-level `base.compile`

14296+ // (before any knot_value_call/knot_io_run) runs its snippet on this handle.

14297+ let _db_scope = RestoreDb::new(db);

14275 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(db))) {

14298 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(db))) {

14276 Ok(v) => v,

14299 Ok(v) => v,

14277 Err(payload) => {

14300 Err(payload) => {

@@ -19694,6 +19717,45 @@ pub extern "C-unwind" fn knot_constraint_register(

19694

19717

19695// ── Atomic (transactions) ─────────────────────────────────────────

19718// ── Atomic (transactions) ─────────────────────────────────────────

19696

19719

19720+thread_local! {

19721+ /// The db handle of the knot code currently executing on this thread —

19722+ /// the handle the innermost active `knot_user_main`/`knot_value_call`/

19723+ /// `knot_io_run` was invoked with. `base.compile` runs its snippet on this

19724+ /// handle (rather than opening a fresh connection) so a compiled value used

19725+ /// inside the host's `atomic` block shares the host's SAVEPOINT: its reads

19726+ /// see uncommitted writes and its own writes roll back with the transaction.

19727+ ///

19728+ /// Set on entry to the foreign-boundary crossings (`knot_run_user_main`,

19729+ /// `knot_value_call`, `knot_io_run`) and restored on return, so a nested

19730+ /// `compile`'s snippet doesn't clobber the host's current-db after the

19731+ /// snippet returns. Null outside any knot execution (e.g. the compiler's

19732+ /// const-eval thread), where `compile` falls back to its own connection.

19733+ static CURRENT_DB: std::cell::Cell<*mut c_void> = const { std::cell::Cell::new(std::ptr::null_mut()) };

19734+}

19735+

19736+/// Set the current-db for this thread, returning the previous value so the

19737+/// caller can restore it on scope exit (see [`RestoreDb`]).

19738+fn set_current_db(db: *mut c_void) -> *mut c_void {

19739+ CURRENT_DB.with(|c| c.replace(db))

19740+}

19741+

19742+/// RAII guard restoring the previous current-db on drop.

19743+struct RestoreDb(*mut c_void);

19744+impl RestoreDb {

19745+ fn new(db: *mut c_void) -> Self {

19746+ RestoreDb(set_current_db(db))

19747+ }

19748+}

19749+impl Drop for RestoreDb {

19750+ fn drop(&mut self) {

19751+ set_current_db(self.0);

19752+ }

19753+}

19754+

19755+/// The current thread's db handle, or null when outside knot execution.

19756+fn current_db() -> *mut c_void {

19757+ CURRENT_DB.with(|c| c.get())

19758+}

19697

19759

19698#[unsafe(no_mangle)]

19760#[unsafe(no_mangle)]

19699pub extern "C-unwind" fn knot_atomic_begin(db: *mut c_void) {

19761pub extern "C-unwind" fn knot_atomic_begin(db: *mut c_void) {