Skip to content

test: give each async launch an owned callback context #49

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
owenthcarey wants to merge 1 commit into
base: main
Choose a base branch
from
Open

test: give each async launch an owned callback context #49

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 40 additions & 34 deletions crates/weaveffi/tests/runtime.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,6 +451,28 @@ fn decode_ret<T: abi::BufferValue>(ptr: *const u8, len: usize) -> T {
value
}

/// Build the `context` for one async launch: a heap-boxed clone of the
/// test's channel sender that the completion callback adopts with
/// [`adopt_ctx`].
///
/// The callback fires on a detached producer thread, and the test's `recv`
/// unblocks as soon as the message is queued, before that thread has finished
/// `send`. Handing the callback a *borrowed* sender and freeing it after
/// `recv` would therefore free the channel out from under an in-flight
/// `send`. Giving every launch its own owned sender, released by the callback
/// itself, means the channel is torn down by whichever side lets go last, the
/// way the ABI's "consumer owns `context`" rule is meant to be used.
fn ctx_for<T>(tx: &std::sync::mpsc::Sender<T>) -> *mut std::os::raw::c_void {
Box::into_raw(Box::new(tx.clone())).cast()
}

/// Take ownership of the sender [`ctx_for`] boxed for this callback. Must be
/// called exactly once per launch, which the "callback fires exactly once"
/// contract guarantees.
unsafe fn adopt_ctx<T>(ctx: *mut std::os::raw::c_void) -> Box<std::sync::mpsc::Sender<T>> {
unsafe { Box::from_raw(ctx.cast()) }
}

#[test]
fn scalar_call_sets_ok() {
let mut err = ok_err();
Expand Down Expand Up @@ -932,7 +954,7 @@ fn callback_interface_from_async_method() {

type Msg = (i32, String, i64);
extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<Msg>) };
let tx = unsafe { adopt_ctx::<Msg>(ctx) };
let (code, msg) = if err.is_null() {
(0, String::new())
} else {
Expand All @@ -950,31 +972,29 @@ fn callback_interface_from_async_method() {
bus::weaveffi_bus_Bus_subscribe(b, new_ctx(4, &freed), &VTABLE, &mut err);

let (tx, rx) = mpsc::channel::<Msg>();
let tx_ptr = Box::into_raw(Box::new(tx));
let text = string_to_c_ptr("async");

bus::weaveffi_bus_Bus_publish_later_async(b, text, 2, cb, tx_ptr as *mut c_void);
bus::weaveffi_bus_Bus_publish_later_async(b, text, 2, cb, ctx_for(&tx));
let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert_eq!(code, 0);
assert_eq!(result, 2);

// A foreign failure inside the future is delivered through the callback.
bus::weaveffi_bus_Bus_publish_later_async(b, text, 4, cb, tx_ptr as *mut c_void);
bus::weaveffi_bus_Bus_publish_later_async(b, text, 4, cb, ctx_for(&tx));
let (code, msg, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert_eq!(code, abi::FOREIGN_ERROR_CODE);
assert_eq!(msg, "subscriber rejected async");
assert_eq!(result, 0);

// The receiver was retained across the spawn: releasing the consumer's
// reference while a call is in flight is safe.
bus::weaveffi_bus_Bus_publish_later_async(b, text, 1, cb, tx_ptr as *mut c_void);
bus::weaveffi_bus_Bus_publish_later_async(b, text, 1, cb, ctx_for(&tx));
bus::weaveffi_bus_Bus_destroy(b);
let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert_eq!(code, 0);
assert_eq!(result, 3);

free_string(text);
unsafe { drop(Box::from_raw(tx_ptr)) };
}

#[test]
Expand All @@ -992,7 +1012,7 @@ fn async_struct_result_completes_via_callback() {
result_ptr: *const u8,
result_len: usize,
) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<Msg>) };
let tx = unsafe { adopt_ctx::<Msg>(ctx) };
let had_err = !err.is_null() && unsafe { (*err).code } != 0;
if had_err {
abi::error_free(err);
Expand All @@ -1009,12 +1029,10 @@ fn async_struct_result_completes_via_callback() {
}

let (tx, rx) = mpsc::channel::<Msg>();
let tx_ptr = Box::into_raw(Box::new(tx));
let name = string_to_c_ptr("alpha");
tasks::weaveffi_tasks_run_task_async(name, cb, tx_ptr as *mut c_void);
tasks::weaveffi_tasks_run_task_async(name, cb, ctx_for(&tx));
let (had_err, id, value) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
free_string(name);
unsafe { drop(Box::from_raw(tx_ptr)) };

assert!(!had_err);
assert_eq!(id, 7);
Expand All @@ -1029,7 +1047,7 @@ fn async_result_ok_and_err_paths() {

type Msg = (bool, i32);
extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i32) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<Msg>) };
let tx = unsafe { adopt_ctx::<Msg>(ctx) };
let had_err = !err.is_null() && unsafe { (*err).code } != 0;
if had_err {
// The reported error is heap-boxed and owned by the consumer.
Expand All @@ -1039,19 +1057,16 @@ fn async_result_ok_and_err_paths() {
}

let (tx, rx) = mpsc::channel::<Msg>();
let tx_ptr = Box::into_raw(Box::new(tx));

tasks::weaveffi_tasks_checked_add_async(2, 3, cb, tx_ptr as *mut c_void);
tasks::weaveffi_tasks_checked_add_async(2, 3, cb, ctx_for(&tx));
let (had_err, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert!(!had_err);
assert_eq!(result, 5);

tasks::weaveffi_tasks_checked_add_async(i32::MAX, 1, cb, tx_ptr as *mut c_void);
tasks::weaveffi_tasks_checked_add_async(i32::MAX, 1, cb, ctx_for(&tx));
let (had_err, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert!(had_err);
assert_eq!(result, 0);

unsafe { drop(Box::from_raw(tx_ptr)) };
}

#[test]
Expand Down Expand Up @@ -1088,7 +1103,7 @@ fn deferred_foreign_error_fires_the_async_callback_once() {

type Msg = (i32, String, i32);
extern "C" fn cb(ctx: *mut c_void, err: *mut weaveffi_error, result: i32) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<Msg>) };
let tx = unsafe { adopt_ctx::<Msg>(ctx) };
let (code, message) = if err.is_null() {
(0, String::new())
} else {
Expand All @@ -1101,13 +1116,12 @@ fn deferred_foreign_error_fires_the_async_callback_once() {
}

let (tx, rx) = mpsc::channel::<Msg>();
let tx_ptr = Box::into_raw(Box::new(tx));

deferred::weaveffi_deferred_later_then_fail_async(false, cb, tx_ptr as *mut c_void);
deferred::weaveffi_deferred_later_then_fail_async(false, cb, ctx_for(&tx));
let (code, _, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert_eq!((code, result), (0, 88));

deferred::weaveffi_deferred_later_then_fail_async(true, cb, tx_ptr as *mut c_void);
deferred::weaveffi_deferred_later_then_fail_async(true, cb, ctx_for(&tx));
let (code, message, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
assert_eq!(code, abi::FOREIGN_ERROR_CODE);
assert_eq!(message, "consumer said no, later");
Expand All @@ -1116,8 +1130,6 @@ fn deferred_foreign_error_fires_the_async_callback_once() {
rx.recv_timeout(Duration::from_millis(200)).is_err(),
"the completion callback fires exactly once"
);

unsafe { drop(Box::from_raw(tx_ptr)) };
}

/// Exercises nested-module codegen: the inner module's symbols must carry the
Expand Down Expand Up @@ -1644,7 +1656,7 @@ fn async_methods_retain_the_receiver() {
use std::time::Duration;

extern "C" fn on_value(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<i64>) };
let tx = unsafe { adopt_ctx::<i64>(ctx) };
assert!(err.is_null());
tx.send(result).unwrap();
}
Expand All @@ -1653,7 +1665,7 @@ fn async_methods_retain_the_receiver() {
err: *mut weaveffi_error,
result: *mut counters::Counter,
) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<i64>) };
let tx = unsafe { adopt_ctx::<i64>(ctx) };
assert!(err.is_null());
let mut e = weaveffi_error::default();
let v = counters::weaveffi_counters_Counter_value(result, &mut e);
Expand All @@ -1663,11 +1675,10 @@ fn async_methods_retain_the_receiver() {

let mut err = ok_err();
let (tx, rx) = mpsc::channel::<i64>();
let tx_ptr = Box::into_raw(Box::new(tx));

let c = counters::weaveffi_counters_Counter_new(8, &mut err);
counters::weaveffi_counters_Counter_value_later_async(c, on_value, tx_ptr as *mut c_void);
counters::weaveffi_counters_Counter_snapshot_later_async(c, on_obj, tx_ptr as *mut c_void);
counters::weaveffi_counters_Counter_value_later_async(c, on_value, ctx_for(&tx));
counters::weaveffi_counters_Counter_snapshot_later_async(c, on_obj, ctx_for(&tx));
// Releasing the consumer's reference while calls are in flight is safe:
// each launcher retained its own.
counters::weaveffi_counters_Counter_destroy(c);
Expand All @@ -1680,19 +1691,14 @@ fn async_methods_retain_the_receiver() {

// A null receiver still completes (with a marshalling error).
extern "C" fn on_null(ctx: *mut c_void, err: *mut weaveffi_error, result: i64) {
let tx = unsafe { &*(ctx as *const mpsc::Sender<i64>) };
let tx = unsafe { adopt_ctx::<i64>(ctx) };
assert!(!err.is_null());
assert_eq!(unsafe { (*err).code }, abi::MARSHAL_ERROR_CODE);
abi::error_free(err);
tx.send(result).unwrap();
}
counters::weaveffi_counters_Counter_value_later_async(
std::ptr::null(),
on_null,
tx_ptr as *mut c_void,
);
counters::weaveffi_counters_Counter_value_later_async(std::ptr::null(), on_null, ctx_for(&tx));
assert_eq!(rx.recv_timeout(Duration::from_secs(5)).unwrap(), 0);
unsafe { drop(Box::from_raw(tx_ptr)) };
}

#[test]
Expand Down
70 changes: 26 additions & 44 deletions samples/async-demo/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ mod tests {
result_ptr: *const u8,
result_len: usize,
) {
let tx = unsafe { &*(context as *const mpsc::Sender<TaskCbMsg>) };
let tx = unsafe { adopt_ctx::<TaskCbMsg>(context) };
let code = if err.is_null() {
0
} else {
Expand All @@ -158,7 +158,7 @@ mod tests {
results_ptr: *const u8,
results_len: usize,
) {
let tx = unsafe { &*(context as *const mpsc::Sender<BatchCbMsg>) };
let tx = unsafe { adopt_ctx::<BatchCbMsg>(context) };
let had_error = !err.is_null() && unsafe { (*err).code } != 0;
// Decode the borrowed value buffer before returning; the launcher
// frees it once the callback completes.
Expand All @@ -172,35 +172,39 @@ mod tests {
}

extern "C" fn n_tasks_callback(context: *mut c_void, err: *mut weaveffi_error, result: i32) {
let tx = unsafe { &*(context as *const mpsc::Sender<(bool, i32)>) };
let tx = unsafe { adopt_ctx::<(bool, i32)>(context) };
let had_error = !err.is_null() && unsafe { (*err).code } != 0;
let _ = tx.send((had_error, result));
}

/// Intentionally leak a callback-context box.
/// Box a clone of the test's sender as the `context` for one async
/// launch, to be adopted by the completion callback with [`adopt_ctx`].
///
/// The `#[weaveffi::module]` async launchers invoke the completion callback
/// on a detached worker thread, so a worker may still be inside the
/// callback's `send` when the test's `recv` returns (the receiver unblocks
/// as soon as the message is queued, before `send` finishes). Reclaiming the
/// box here would free the `Sender` out from under that in-flight `send`, a
/// use-after-free. The test deliberately leaks the context instead, which
/// keeps the channel alive for the brief remaining life of any in-flight
/// callback; the OS reclaims the memory at process exit.
fn leak_ctx<T>(ptr: *mut T) {
std::mem::forget(unsafe { Box::from_raw(ptr) });
/// on a detached worker thread, and the test's `recv` unblocks as soon as
/// the message is queued, before that worker has finished `send`. Handing
/// the callback a borrowed sender and freeing it after `recv` would
/// therefore free the channel out from under an in-flight `send`. Giving
/// each launch its own owned sender, released by the callback itself,
/// means the channel is torn down by whichever side lets go last.
fn ctx_for<T>(tx: &mpsc::Sender<T>) -> *mut c_void {
Box::into_raw(Box::new(tx.clone())).cast()
}

/// Take ownership of the sender [`ctx_for`] boxed for this callback. The
/// "callback fires exactly once" contract makes this a single adoption.
unsafe fn adopt_ctx<T>(ctx: *mut c_void) -> Box<mpsc::Sender<T>> {
unsafe { Box::from_raw(ctx.cast()) }
}

#[test]
fn run_task_calls_callback() {
let (tx, rx) = mpsc::channel::<TaskCbMsg>();
let tx_ptr = Box::into_raw(Box::new(tx));
let name = CString::new("test-task").unwrap();

weaveffi_tasks_run_task_async(name.as_ptr(), task_callback, tx_ptr as *mut c_void);
weaveffi_tasks_run_task_async(name.as_ptr(), task_callback, ctx_for(&tx));

let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
leak_ctx(tx_ptr);
assert_eq!(code, 0);

let r = result.expect("success path passes a result buffer");
Expand All @@ -212,37 +216,32 @@ mod tests {
#[test]
fn run_task_empty_name_reports_invalid_name() {
let (tx, rx) = mpsc::channel::<TaskCbMsg>();
let tx_ptr = Box::into_raw(Box::new(tx));

let empty = CString::new("").unwrap();
weaveffi_tasks_run_task_async(empty.as_ptr(), task_callback, tx_ptr as *mut c_void);
weaveffi_tasks_run_task_async(empty.as_ptr(), task_callback, ctx_for(&tx));

let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
leak_ctx(tx_ptr);
assert_eq!(code, 1, "TaskError::InvalidName's declared code");
assert!(result.is_none(), "error path passes a null result buffer");
}

#[test]
fn run_task_null_name_reports_marshal_error_through_the_callback() {
let (tx, rx) = mpsc::channel::<TaskCbMsg>();
let tx_ptr = Box::into_raw(Box::new(tx));

// The launcher has no out_err slot, so an argument that fails to lift
// is reported through the completion callback with the reserved
// marshalling code, exactly like the sync path would report it.
weaveffi_tasks_run_task_async(std::ptr::null(), task_callback, tx_ptr as *mut c_void);
weaveffi_tasks_run_task_async(std::ptr::null(), task_callback, ctx_for(&tx));

let (code, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
leak_ctx(tx_ptr);
assert_eq!(code, abi::MARSHAL_ERROR_CODE);
assert!(result.is_none());
}

#[test]
fn run_batch_processes_sequentially() {
let (tx, rx) = mpsc::channel::<BatchCbMsg>();
let tx_ptr = Box::into_raw(Box::new(tx));
// The list-of-strings parameter is buffered: encode `Vec<String>`
// and pass the (ptr, len) pair. The launcher copies the bytes before
// returning, so the local buffer only needs to outlive the call.
Expand All @@ -252,15 +251,9 @@ mod tests {
"task-c".to_string(),
]);

weaveffi_tasks_run_batch_async(
names.as_ptr(),
names.len(),
batch_callback,
tx_ptr as *mut c_void,
);
weaveffi_tasks_run_batch_async(names.as_ptr(), names.len(), batch_callback, ctx_for(&tx));

let (had_error, results) = rx.recv_timeout(Duration::from_secs(10)).unwrap();
leak_ctx(tx_ptr);
assert!(!had_error);
assert_eq!(results.len(), 3);

Expand All @@ -276,18 +269,11 @@ mod tests {
#[test]
fn run_batch_empty_names() {
let (tx, rx) = mpsc::channel::<BatchCbMsg>();
let tx_ptr = Box::into_raw(Box::new(tx));

let names = abi::encode_value(&Vec::<String>::new());
weaveffi_tasks_run_batch_async(
names.as_ptr(),
names.len(),
batch_callback,
tx_ptr as *mut c_void,
);
weaveffi_tasks_run_batch_async(names.as_ptr(), names.len(), batch_callback, ctx_for(&tx));

let (had_error, results) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
leak_ctx(tx_ptr);
assert!(!had_error);
assert!(results.is_empty());
}
Expand Down Expand Up @@ -331,10 +317,8 @@ mod tests {
#[test]
fn run_n_tasks_invokes_callback_with_n() {
let (tx, rx) = mpsc::channel::<(bool, i32)>();
let tx_ptr = Box::into_raw(Box::new(tx));
weaveffi_tasks_run_n_tasks_async(7, n_tasks_callback, tx_ptr as *mut c_void);
weaveffi_tasks_run_n_tasks_async(7, n_tasks_callback, ctx_for(&tx));
let (had_error, result) = rx.recv_timeout(Duration::from_secs(5)).unwrap();
leak_ctx(tx_ptr);
assert!(!had_error);
assert_eq!(result, 7);
}
Expand All @@ -343,15 +327,13 @@ mod tests {
fn active_callbacks_returns_to_zero() {
let mut err = weaveffi_error::default();
let (tx, rx) = mpsc::channel::<(bool, i32)>();
let tx_ptr = Box::into_raw(Box::new(tx));

for i in 0..16 {
weaveffi_tasks_run_n_tasks_async(i, n_tasks_callback, tx_ptr as *mut c_void);
weaveffi_tasks_run_n_tasks_async(i, n_tasks_callback, ctx_for(&tx));
}
for _ in 0..16 {
rx.recv_timeout(Duration::from_secs(5)).unwrap();
}
leak_ctx(tx_ptr);

for _ in 0..50 {
if weaveffi_tasks_active_callbacks(&mut err) == 0 {
Expand Down
Loading
Loading