fix(zset-ffi): first fault wins, no free or re-entry under a running step, circuit ids, interrupted rebuilds, typed callbacks (improvise-bw4p.1)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze
This commit is contained in:
committed by
Edward Langley
parent
a23851e920
commit
42d808732b
@@ -62,7 +62,12 @@ link, so they call the `extern "C"` functions as plain functions.
|
||||
duration of `step` and `rebuild_step`, and cleared only when they return. Every entry
|
||||
point first reads it, so a circuit whose frames were skipped answers `ZS_ERR_POISONED`
|
||||
forever, and a callback that re-enters its own circuit gets `ZS_ERR_POISONED` without the
|
||||
library touching the engine. The CL rule is to put `handler-case` *inside* the callback and
|
||||
library touching the engine. `zs_rebuild_step` reads `circuit(c)` before it borrows the job,
|
||||
so re-entry from an `expired` callback is refused first. `zs_circuit_free` on a mid-step
|
||||
circuit and `zs_rebuild_free` on a job being stepped do nothing: the object is leaked, not
|
||||
freed under the frames that are running. The first fault recorded in a step is the one
|
||||
reported (`note`), so a failing `index_by` callback is `ZS_ERR_CALLBACK`, not the
|
||||
consequent `BAD_VALUE`. The CL rule is to put `handler-case` *inside* the callback and
|
||||
return the failure value (`tests/smoke.lisp` does). `index_by` needs exactly one emitted
|
||||
key and a keyed stream valid pairs, else `ZS_ERR_BAD_VALUE`; these faults are recorded in a
|
||||
shared cell rather than panicking.
|
||||
@@ -87,8 +92,9 @@ last chunk ran, whatever the deadline callback says.
|
||||
All exported functions are `pub unsafe extern "C" fn`: the rlib is public API, and a safe
|
||||
Rust caller could otherwise pass any pointer. Output pointers are checked before any work
|
||||
(`with_out`): a refused call changes nothing and leaks nothing. Stream ids and sink ids are
|
||||
different spaces (sink ids carry the top bit). A `ZsRebuild` records its circuit and is
|
||||
refused on another. An id from another circuit is not detected. `zs_zset_add(a, a)` is
|
||||
different spaces (sink ids carry the top bit). A `ZsRebuild` records its circuit's id (a process-wide counter, not an address, which can be
|
||||
reused) and is refused on another. A `Circuit::step` between two `zs_rebuild_step` calls is
|
||||
`ZS_ERR_INTERRUPTED`, without poisoning: zset's `Rebuild::interrupted`, checked first. An id from another circuit is not detected. `zs_zset_add(a, a)` is
|
||||
defined: the sum is computed through shared references and then stored. Keys of 4 GiB or
|
||||
longer are `ZS_ERR_BAD_VALUE`, not truncated. A non-empty aggregate group always has a row,
|
||||
even when its count or sum is zero.
|
||||
@@ -96,14 +102,17 @@ even when its count or sum is zero.
|
||||
## Miri
|
||||
|
||||
Miri checks the aliasing, leak and null rules the C compiler cannot. It runs `tests/abi.rs`
|
||||
on a nightly built through the flake's `rust-overlay` input (passes at this commit, 12s):
|
||||
on a nightly built through the flake's `rust-overlay` input. The recipe below is **not
|
||||
pinned**: `builtins.getFlake "github:oxalica/rust-overlay"` and `<nixpkgs>` (hence `--impure`)
|
||||
resolve to whatever is current, and only the nightly date is fixed. It passed at the date of
|
||||
this commit (about 12s):
|
||||
|
||||
```
|
||||
nix shell --impure --expr 'let p = import <nixpkgs> { overlays = [ (builtins.getFlake "github:oxalica/rust-overlay").overlays.default ]; }; in p.rust-bin.nightly."2026-09-24".default.override { extensions = [ "miri" "rust-src" ]; }' --command cargo miri test -p zset-ffi --test abi
|
||||
nix shell --impure --expr 'let p = import <nixpkgs> { overlays = [ (builtins.getFlake "github:oxalica/rust-overlay").overlays.default ]; }; in p.rust-bin.nightly."2026-09-24".default.override { extensions = [ "miri" "rust-src" ]; }' --command cargo miri test -p zset-ffi --lib --test abi
|
||||
```
|
||||
|
||||
It is not a CI job. A proposed one: a `miri` job in `.github/workflows/ci.yml` running that
|
||||
command in `nix develop .#ci`.
|
||||
It is not a CI job. A proposed one: a `miri` job in `.github/workflows/ci.yml` that takes the
|
||||
nightly from the flake's own pinned `rust-overlay` input in `nix develop .#ci`.
|
||||
|
||||
## Calling it from Common Lisp
|
||||
|
||||
|
||||
@@ -132,3 +132,37 @@ Codex's finding: `Some(0)` was filtered away for a non-empty group.
|
||||
### a_length_of_4_gib_or_more_is_refused_not_truncated
|
||||
|
||||
`len32` refuses `1 << 32` and accepts `u32::MAX`.
|
||||
|
||||
## Delta review of #347 (9bcfa9d1)
|
||||
|
||||
**Injected**, one at a time: a fault that overwrites the first (`note` replaced by a plain
|
||||
`set`) fails `an_index_by_callback_that_fails_reports_the_callback_code_not_a_later_fault`;
|
||||
`zs_rebuild_step` borrowing the job before checking the circuit (Miri only), and a `free` that ignores
|
||||
`in_step` or `busy`, fail `a_callback_cannot_step_or_free_the_job_or_circuit_it_runs_under`
|
||||
(and under Miri, as use-after-free); an owner compared by address instead of id fails
|
||||
`a_circuit_freed_and_another_made_does_not_inherit_its_jobs`; the interrupted check removed
|
||||
fails `a_host_step_between_rebuild_calls_is_interrupted_not_a_poison`; a retyped
|
||||
`ZsKeepFn` fails `the_header_callback_typedefs_match_the_rust_callback_types`.
|
||||
|
||||
### an_index_by_callback_that_fails_reports_the_callback_code_not_a_later_fault
|
||||
|
||||
The callback fails (returns 3) and emits nothing, so `index_by` also sees zero keys. The step
|
||||
must report `ZS_ERR_CALLBACK`, the first fault, not the consequent `BAD_VALUE`.
|
||||
|
||||
### a_callback_cannot_step_or_free_the_job_or_circuit_it_runs_under
|
||||
|
||||
An `expired` callback steps its own job, frees the job and frees the circuit. The step is
|
||||
refused with `ZS_ERR_POISONED`, both frees do nothing, and the job then finishes normally.
|
||||
|
||||
### a_circuit_freed_and_another_made_does_not_inherit_its_jobs
|
||||
|
||||
A job outlives its circuit; sixty-four new circuits (some at the old address) all refuse it.
|
||||
|
||||
### a_host_step_between_rebuild_calls_is_interrupted_not_a_poison
|
||||
|
||||
`ZS_ERR_INTERRUPTED`, and the circuit still steps afterwards.
|
||||
|
||||
### the_header_callback_typedefs_match_the_rust_callback_types
|
||||
|
||||
Retyping `ZsKeepFn`'s return in the header fails. The `unary_op!` signatures are now read
|
||||
from the macro body rather than written into the test.
|
||||
|
||||
@@ -21,8 +21,8 @@
|
||||
* a reference count): keep the whole group on one thread at a time. Z-sets are separate
|
||||
* handles. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is
|
||||
* never a stream id (they are refused as ZS_ERR_BAD_HANDLE), but an id from another circuit
|
||||
* is not detected. A ZsRebuild is stepped only on its own circuit (else ZS_ERR_BAD_HANDLE);
|
||||
* free the jobs before the circuit.
|
||||
* is not detected. A ZsRebuild is stepped only on the circuit that made it (else
|
||||
* ZS_ERR_BAD_HANDLE), even if another circuit now lives at the old address.
|
||||
* - Keys of 4 GiB or longer are refused with ZS_ERR_BAD_VALUE: the wire form has u32 lengths.
|
||||
*/
|
||||
#ifndef ZSET_FFI_H
|
||||
@@ -46,6 +46,7 @@ extern "C" {
|
||||
#define ZS_ERR_BAD_VALUE 6
|
||||
#define ZS_ERR_POISONED 7
|
||||
#define ZS_ERR_CALLBACK 8
|
||||
#define ZS_ERR_INTERRUPTED 9
|
||||
|
||||
#define ZS_AGG_COUNT 0
|
||||
#define ZS_AGG_SUM 1
|
||||
@@ -66,8 +67,11 @@ typedef struct ZsRebuild ZsRebuild;
|
||||
* - A callback MUST return normally: no longjmp, no C++ exception, no Lisp non-local exit
|
||||
* through it (wrap the body in handler-case and return the failure value). If one does,
|
||||
* the circuit is left marked mid-step and every later call on it is ZS_ERR_POISONED.
|
||||
* - A callback must not call back into the library for its own circuit: that call returns
|
||||
* ZS_ERR_POISONED. */
|
||||
* - A callback that calls the library with its own circuit (or the rebuild job being
|
||||
* stepped) gets ZS_ERR_POISONED and nothing changes. zs_circuit_free on a circuit that is
|
||||
* mid-step, and zs_rebuild_free on a job being stepped, do nothing (the object is leaked
|
||||
* rather than freed under the running frames); free after the call returns.
|
||||
* A circuit whose step was left by a non-local exit stays mid-step, so its free leaks too. */
|
||||
typedef void (*ZsEmit)(void *emit_ctx, const uint8_t *bytes, size_t len);
|
||||
typedef int32_t (*ZsFlatMapFn)(void *ctx, const uint8_t *key, size_t len, ZsEmit emit, void *emit_ctx);
|
||||
/* 1 keep, 0 drop, negative fail. */
|
||||
@@ -132,7 +136,9 @@ int32_t zs_circuit_step(ZsCircuit *c);
|
||||
int32_t zs_circuit_take(ZsCircuit *c, uint32_t sink, ZsZSet **out);
|
||||
|
||||
/* Resumable snapshot load: chunk keys per piece (0 reads as 1). The whole load is one
|
||||
* logical tick. done is 1 as soon as the last chunk ran. */
|
||||
* logical tick, and done is 1 as soon as the last chunk ran. While a job is pending it
|
||||
* owns the clock: a zs_circuit_step in between makes the next zs_rebuild_step return
|
||||
* ZS_ERR_INTERRUPTED (the circuit stays usable; drop the job). */
|
||||
int32_t zs_rebuild_new(ZsCircuit *c, uint32_t input, const ZsZSet *snapshot, size_t chunk, ZsRebuild **out);
|
||||
int32_t zs_rebuild_step(ZsRebuild *job, ZsCircuit *c, ZsExpiredFn expired, void *ctx, int32_t *done);
|
||||
void zs_rebuild_free(ZsRebuild *job);
|
||||
|
||||
@@ -39,6 +39,7 @@ codes! {
|
||||
BadValue = 6 => "ZS_ERR_BAD_VALUE",
|
||||
Poisoned = 7 => "ZS_ERR_POISONED",
|
||||
Callback = 8 => "ZS_ERR_CALLBACK",
|
||||
Interrupted = 9 => "ZS_ERR_INTERRUPTED",
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
@@ -134,6 +135,15 @@ struct Ctx(*mut c_void);
|
||||
|
||||
type Fault = Rc<Cell<Option<(Code, &'static str)>>>;
|
||||
|
||||
/// Records a fault unless one is already recorded: the first one is the cause.
|
||||
fn note(fault: &Fault, code: Code, msg: &'static str) {
|
||||
if let Some(first) = fault.take() {
|
||||
fault.set(Some(first));
|
||||
} else {
|
||||
fault.set(Some((code, msg)));
|
||||
}
|
||||
}
|
||||
|
||||
extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) {
|
||||
// SAFETY: `ctx` is the `Vec` `images` passes below, alive for the callback's duration.
|
||||
let out = unsafe { &mut *(ctx as *mut Vec<Bytes>) };
|
||||
@@ -157,7 +167,7 @@ fn images(f: FlatMapFn, ctx: Ctx, key: &Bytes, fault: &Fault) -> Vec<Bytes> {
|
||||
&mut out as *mut Vec<Bytes> as *mut c_void,
|
||||
);
|
||||
if rc != 0 {
|
||||
fault.set(Some((Code::Callback, "a callback reported failure")));
|
||||
note(fault, Code::Callback, "a callback reported failure");
|
||||
out.clear();
|
||||
}
|
||||
out
|
||||
@@ -299,7 +309,7 @@ impl Engine {
|
||||
0 => false,
|
||||
n if n > 0 => true,
|
||||
_ => {
|
||||
fault.set(Some((Code::Callback, "a callback reported failure")));
|
||||
note(&fault, Code::Callback, "a callback reported failure");
|
||||
false
|
||||
}
|
||||
}
|
||||
@@ -325,16 +335,17 @@ impl Engine {
|
||||
let out = self.circuit.flat_map(&s, move |k: &Bytes| {
|
||||
let mut gs = images(f, ctx, k, &fault);
|
||||
if gs.len() != 1 {
|
||||
fault.set(Some((
|
||||
note(
|
||||
&fault,
|
||||
Code::BadValue,
|
||||
"index_by: the key callback must emit exactly one key",
|
||||
)));
|
||||
);
|
||||
return Vec::new();
|
||||
}
|
||||
match pair(&gs.remove(0), k) {
|
||||
Ok(p) => vec![p],
|
||||
Err(_) => {
|
||||
fault.set(Some((Code::BadValue, "a key is 4 GiB or longer")));
|
||||
note(&fault, Code::BadValue, "a key is 4 GiB or longer");
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
@@ -348,10 +359,11 @@ impl Engine {
|
||||
self.circuit.flat_map(s, move |b: &Bytes| match unpair(b) {
|
||||
Some((k, v)) => vec![(k.to_vec(), v.to_vec())],
|
||||
None => {
|
||||
fault.set(Some((
|
||||
note(
|
||||
&fault,
|
||||
Code::BadValue,
|
||||
"a keyed stream carried an element that is not a pair",
|
||||
)));
|
||||
);
|
||||
Vec::new()
|
||||
}
|
||||
})
|
||||
@@ -367,7 +379,7 @@ impl Engine {
|
||||
.circuit
|
||||
.join_indexed(&ka, &kb, move |k: &Bytes, x: &Bytes, y: &Bytes| {
|
||||
pair(x, y).and_then(|xy| pair(k, &xy)).unwrap_or_else(|_| {
|
||||
fault.set(Some((Code::BadValue, "a key is 4 GiB or longer")));
|
||||
note(&fault, Code::BadValue, "a key is 4 GiB or longer");
|
||||
Vec::new()
|
||||
})
|
||||
});
|
||||
@@ -393,7 +405,7 @@ impl Engine {
|
||||
match <[u8; 8]>::try_from(v.as_slice()) {
|
||||
Ok(b) => nums.push((i64::from_le_bytes(b), *w)),
|
||||
Err(_) => {
|
||||
fault.set(Some((Code::BadValue, "an aggregated value is not 8 bytes")));
|
||||
note(&fault, Code::BadValue, "an aggregated value is not 8 bytes");
|
||||
return None;
|
||||
}
|
||||
}
|
||||
@@ -418,7 +430,7 @@ impl Engine {
|
||||
match pair(k, &r.to_le_bytes()) {
|
||||
Ok(p) => vec![p],
|
||||
Err(_) => {
|
||||
fault.set(Some((Code::BadValue, "a key is 4 GiB or longer")));
|
||||
note(&fault, Code::BadValue, "a key is 4 GiB or longer");
|
||||
Vec::new()
|
||||
}
|
||||
}
|
||||
@@ -482,6 +494,13 @@ impl Engine {
|
||||
job: &mut Rebuild<Bytes>,
|
||||
expired: Option<(ExpiredFn, *mut c_void)>,
|
||||
) -> Res<bool> {
|
||||
self.alive()?;
|
||||
if job.interrupted(&self.circuit) {
|
||||
return Err(Fail::new(
|
||||
Code::Interrupted,
|
||||
"the rebuild was interrupted by a circuit step before it finished",
|
||||
));
|
||||
}
|
||||
let fault = Rc::clone(&self.fault);
|
||||
let progress = self.guarded(|c| {
|
||||
job.step(c, || match expired {
|
||||
@@ -490,7 +509,7 @@ impl Engine {
|
||||
0 => false,
|
||||
n if n > 0 => true,
|
||||
_ => {
|
||||
fault.set(Some((Code::Callback, "a callback reported failure")));
|
||||
note(&fault, Code::Callback, "a callback reported failure");
|
||||
true
|
||||
}
|
||||
},
|
||||
|
||||
@@ -9,6 +9,7 @@ use engine::{
|
||||
use std::cell::{Cell, RefCell, UnsafeCell};
|
||||
use std::ffi::{CString, c_char, c_void};
|
||||
use std::panic::{AssertUnwindSafe, catch_unwind};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use zset::{Rebuild, ZSet};
|
||||
|
||||
pub const ZS_ABI_VERSION: u32 = 1;
|
||||
@@ -21,12 +22,16 @@ pub struct ZsZSet(BZ);
|
||||
pub struct ZsCircuit {
|
||||
engine: UnsafeCell<Engine>,
|
||||
in_step: Cell<bool>,
|
||||
id: u64,
|
||||
}
|
||||
|
||||
static NEXT_CIRCUIT_ID: AtomicU64 = AtomicU64::new(1);
|
||||
|
||||
/// An opaque resumable snapshot load, bound to the circuit that made it.
|
||||
pub struct ZsRebuild {
|
||||
job: Rebuild<Bytes>,
|
||||
owner: *const ZsCircuit,
|
||||
owner: u64,
|
||||
busy: Cell<bool>,
|
||||
}
|
||||
|
||||
thread_local! {
|
||||
@@ -318,6 +323,7 @@ pub unsafe extern "C" fn zs_circuit_new(out: *mut *mut ZsCircuit) -> i32 {
|
||||
boxed(ZsCircuit {
|
||||
engine: UnsafeCell::new(Engine::new()),
|
||||
in_step: Cell::new(false),
|
||||
id: NEXT_CIRCUIT_ID.fetch_add(1, Ordering::Relaxed),
|
||||
})
|
||||
})
|
||||
}
|
||||
@@ -328,7 +334,10 @@ pub unsafe extern "C" fn zs_circuit_new(out: *mut *mut ZsCircuit) -> i32 {
|
||||
/// `c` is null, or a live circuit freed once and not mid-step.
|
||||
#[unsafe(no_mangle)]
|
||||
pub unsafe extern "C" fn zs_circuit_free(c: *mut ZsCircuit) {
|
||||
// SAFETY: the caller's contract.
|
||||
// SAFETY: the caller's contract; a mid-step circuit is leaked, never freed under its frames.
|
||||
if !c.is_null() && unsafe { &*c }.in_step.get() {
|
||||
return;
|
||||
}
|
||||
unsafe { release(c) }
|
||||
}
|
||||
|
||||
@@ -520,7 +529,11 @@ pub unsafe extern "C" fn zs_rebuild_new(
|
||||
with_out(out, || {
|
||||
let (cr, snap) = (circuit(c)?, as_ref(snapshot)?);
|
||||
let job = cr.engine().start_rebuild(input, snap.0.clone(), chunk)?;
|
||||
boxed(ZsRebuild { job, owner: c })
|
||||
boxed(ZsRebuild {
|
||||
job,
|
||||
owner: cr.id,
|
||||
busy: Cell::new(false),
|
||||
})
|
||||
})
|
||||
}
|
||||
})
|
||||
@@ -540,15 +553,23 @@ pub unsafe extern "C" fn zs_rebuild_step(
|
||||
// SAFETY: the caller's contract.
|
||||
unsafe {
|
||||
with_out(done, || {
|
||||
let (job, cr) = (as_mut(job)?, circuit(c)?);
|
||||
if !std::ptr::eq(job.owner, c) {
|
||||
let cr = circuit(c)?;
|
||||
if job.is_null() {
|
||||
return Err(null());
|
||||
}
|
||||
// Field places only, never a `&mut ZsRebuild`: a callback reads `busy` through
|
||||
// `zs_rebuild_free` while `job.job` is mutably borrowed here.
|
||||
if (*job).owner != cr.id {
|
||||
return Err(Fail::new(
|
||||
Code::BadHandle,
|
||||
"the rebuild belongs to another circuit",
|
||||
));
|
||||
}
|
||||
(*job).busy.set(true);
|
||||
let finished =
|
||||
cr.stepping(|e| e.rebuild_step(&mut job.job, expired.map(|f| (f, ctx))))?;
|
||||
cr.stepping(|e| e.rebuild_step(&mut (*job).job, expired.map(|f| (f, ctx))));
|
||||
(*job).busy.set(false);
|
||||
let finished = finished?;
|
||||
Ok(i32::from(finished))
|
||||
})
|
||||
}
|
||||
@@ -559,7 +580,10 @@ pub unsafe extern "C" fn zs_rebuild_step(
|
||||
/// `job` is null, or a live job freed once.
|
||||
#[unsafe(no_mangle)]
|
||||
pub unsafe extern "C" fn zs_rebuild_free(job: *mut ZsRebuild) {
|
||||
// SAFETY: the caller's contract.
|
||||
// SAFETY: the caller's contract; a job being stepped is leaked, not freed under its frame.
|
||||
if !job.is_null() && unsafe { (*job).busy.get() } {
|
||||
return;
|
||||
}
|
||||
unsafe { release(job) }
|
||||
}
|
||||
|
||||
@@ -582,6 +606,7 @@ mod tests {
|
||||
Code::Poisoned as i32
|
||||
);
|
||||
assert_eq!(id, 77, "nothing was written");
|
||||
unsafe { &*c }.in_step.set(false);
|
||||
unsafe { zs_circuit_free(c) };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -834,3 +834,134 @@ fn a_length_of_4_gib_or_more_is_refused_not_truncated() {
|
||||
Code::BadValue
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_index_by_callback_that_fails_reports_the_callback_code_not_a_later_fault() {
|
||||
extern "C" fn fails_after_emitting(
|
||||
_: *mut c_void,
|
||||
_: *const u8,
|
||||
_: usize,
|
||||
_: engine::Emit,
|
||||
_: *mut c_void,
|
||||
) -> i32 {
|
||||
3
|
||||
}
|
||||
let c = C::new();
|
||||
let i = c.input();
|
||||
let mut o = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_index_by(c.0, i, Some(fails_after_emitting), null_mut(), &mut o) },
|
||||
OK
|
||||
);
|
||||
c.push(i, &[(b"x", 1)]);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
|
||||
}
|
||||
|
||||
struct Reenter {
|
||||
c: *mut ZsCircuit,
|
||||
job: *mut ZsRebuild,
|
||||
seen: Vec<i32>,
|
||||
}
|
||||
|
||||
extern "C" fn expired_reenters(ctx: *mut c_void) -> i32 {
|
||||
let r = unsafe { &mut *(ctx as *mut Reenter) };
|
||||
let mut done = 0;
|
||||
r.seen
|
||||
.push(unsafe { zs_rebuild_step(r.job, r.c, None, null_mut(), &mut done) });
|
||||
unsafe { zs_rebuild_free(r.job) };
|
||||
unsafe { zs_circuit_free(r.c) };
|
||||
1
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_callback_cannot_step_or_free_the_job_or_circuit_it_runs_under() {
|
||||
let snapshot = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
|
||||
let c = C::new();
|
||||
let i = c.input();
|
||||
let s = c.sink(i);
|
||||
let mut job = null_mut();
|
||||
assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
|
||||
let mut r = Reenter {
|
||||
c: c.0,
|
||||
job,
|
||||
seen: Vec::new(),
|
||||
};
|
||||
let mut done = 0;
|
||||
let ctx = (&mut r as *mut Reenter).cast();
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c.0, Some(expired_reenters), ctx, &mut done) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(
|
||||
r.seen,
|
||||
vec![Code::Poisoned as i32],
|
||||
"stepping from inside is refused"
|
||||
);
|
||||
assert_eq!(done, 0);
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(
|
||||
(done, c.take(s).len()),
|
||||
(1, 3),
|
||||
"the job and circuit survived the frees"
|
||||
);
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snapshot) };
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_circuit_freed_and_another_made_does_not_inherit_its_jobs() {
|
||||
let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
|
||||
let c1 = C::new();
|
||||
let i1 = c1.input();
|
||||
let mut job = null_mut();
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_new(c1.0, i1, snapshot, 1, &mut job) },
|
||||
OK
|
||||
);
|
||||
drop(c1);
|
||||
for _ in 0..64 {
|
||||
let c2 = C::new();
|
||||
let _ = c2.input();
|
||||
let mut done = 0;
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c2.0, None, null_mut(), &mut done) },
|
||||
Code::BadHandle as i32,
|
||||
"a new circuit, even at the old address, is not the job's"
|
||||
);
|
||||
}
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snapshot) };
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
|
||||
let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
|
||||
let c = C::new();
|
||||
let i = c.input();
|
||||
let s = c.sink(i);
|
||||
let mut job = null_mut();
|
||||
assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
|
||||
let (mut calls, mut done) = (0u32, 0);
|
||||
let ctx = (&mut calls as *mut u32).cast();
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
|
||||
OK
|
||||
);
|
||||
assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
|
||||
assert_eq!(
|
||||
unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
|
||||
Code::Interrupted as i32
|
||||
);
|
||||
assert!(last_error().contains("interrupted"), "{}", last_error());
|
||||
assert_eq!(
|
||||
unsafe { zs_circuit_step(c.0) },
|
||||
OK,
|
||||
"the circuit is still usable"
|
||||
);
|
||||
let _ = c.take(s);
|
||||
unsafe { zs_rebuild_free(job) };
|
||||
unsafe { zs_zset_free(snapshot) };
|
||||
}
|
||||
|
||||
@@ -60,6 +60,7 @@ fn c_type(rust: &str) -> String {
|
||||
"Option<KeepFn>" => "ZsKeepFn",
|
||||
"Option<FlatMapFn>" => "ZsFlatMapFn",
|
||||
"Option<ExpiredFn>" => "ZsExpiredFn",
|
||||
"Emit" => "ZsEmit",
|
||||
other => panic!("no C type is declared for the Rust type `{other}`"),
|
||||
};
|
||||
squash(c)
|
||||
@@ -95,13 +96,24 @@ fn exported() -> BTreeMap<String, (String, Vec<String>)> {
|
||||
};
|
||||
out.insert(name, (ret, params));
|
||||
}
|
||||
// The `unary_op!` expansions all take (circuit, stream, out).
|
||||
// The `unary_op!` expansions share the macro body's signature.
|
||||
let body = &SRC[SRC.find("macro_rules! unary_op").unwrap()..];
|
||||
let sig = body.split("extern \"C\" fn $name(").nth(1).unwrap();
|
||||
let (params, tail) = sig.split_once(')').unwrap();
|
||||
let unary_params: Vec<String> = split_params(params)
|
||||
.iter()
|
||||
.map(|p| c_type(p.split_once(':').unwrap().1))
|
||||
.collect();
|
||||
let unary_ret = c_type(
|
||||
tail.split('{')
|
||||
.next()
|
||||
.unwrap()
|
||||
.trim()
|
||||
.trim_start_matches("->"),
|
||||
);
|
||||
for line in SRC.lines().filter(|l| l.starts_with("unary_op!(")) {
|
||||
let name = line["unary_op!(".len()..].split(',').next().unwrap();
|
||||
let params = ["ZsCircuit*", "uint32_t", "uint32_t*"]
|
||||
.map(String::from)
|
||||
.to_vec();
|
||||
out.insert(name.to_string(), ("int32_t".to_string(), params));
|
||||
out.insert(name.to_string(), (unary_ret.clone(), unary_params.clone()));
|
||||
}
|
||||
out
|
||||
}
|
||||
@@ -202,3 +214,65 @@ fn the_header_constants_are_the_rust_constants() {
|
||||
[0, 1, 2, 3]
|
||||
);
|
||||
}
|
||||
|
||||
/// `typedef R (*Name)(params)` from the header, and `pub type X = extern "C" fn(..) -> R` from
|
||||
/// `engine.rs`, as `name -> (return, params)`.
|
||||
type Sigs = BTreeMap<String, (String, Vec<String>)>;
|
||||
|
||||
fn callback_types() -> (Sigs, Sigs) {
|
||||
let mut hdr = BTreeMap::new();
|
||||
for decl in strip_comments(HEADER).split(';') {
|
||||
let Some(at) = decl.find("(*") else { continue };
|
||||
if !decl.contains("typedef") {
|
||||
continue;
|
||||
}
|
||||
let ret = squash(
|
||||
decl.split("typedef")
|
||||
.nth(1)
|
||||
.unwrap()
|
||||
.split('(')
|
||||
.next()
|
||||
.unwrap(),
|
||||
);
|
||||
let rest = &decl[at + 2..];
|
||||
let (name, after) = rest.split_once(')').unwrap();
|
||||
let params = after
|
||||
.trim()
|
||||
.trim_start_matches('(')
|
||||
.rsplit_once(')')
|
||||
.unwrap()
|
||||
.0;
|
||||
let params = split_params(params).iter().map(|p| param_type(p)).collect();
|
||||
hdr.insert(name.trim().to_string(), (ret, params));
|
||||
}
|
||||
let engine = include_str!("../src/engine.rs");
|
||||
let mut rust = BTreeMap::new();
|
||||
for line in engine.lines().filter(|l| l.contains("= extern \"C\" fn(")) {
|
||||
let name = line
|
||||
.trim_start_matches("pub type ")
|
||||
.split(' ')
|
||||
.next()
|
||||
.unwrap();
|
||||
let sig = line.split("fn(").nth(1).unwrap();
|
||||
let (params, tail) = sig.rsplit_once(')').unwrap();
|
||||
let ret = match tail.trim().trim_end_matches(';').trim().strip_prefix("->") {
|
||||
Some(r) => c_type(r),
|
||||
None => "void".to_string(),
|
||||
};
|
||||
let params = split_params(params).iter().map(|p| c_type(p)).collect();
|
||||
rust.insert(format!("Zs{name}"), (ret, params));
|
||||
}
|
||||
(hdr, rust)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_header_callback_typedefs_match_the_rust_callback_types() {
|
||||
let (hdr, rust) = callback_types();
|
||||
assert_eq!(
|
||||
rust.len(),
|
||||
4,
|
||||
"the scanner saw {} callback types",
|
||||
rust.len()
|
||||
);
|
||||
assert_eq!(rust, hdr);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user