#![doc = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/docs/runtime/async_handler.md"))]
use crate::core::address::Address;
use crate::core::distribution::Distribution;
use crate::core::model::Model;
use crate::runtime::handler::Handler;
use crate::runtime::trace::Trace;
#[allow(async_fn_in_trait)]
pub trait AsyncHandler {
async fn on_sample_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>) -> f64;
async fn on_sample_bool(&mut self, addr: &Address, dist: &dyn Distribution<bool>) -> bool;
async fn on_sample_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>) -> u64;
async fn on_sample_usize(&mut self, addr: &Address, dist: &dyn Distribution<usize>) -> usize;
async fn on_sample_i64(&mut self, addr: &Address, _dist: &dyn Distribution<i64>) -> i64 {
panic!(
"handler does not implement on_sample_i64 (i64 sample site at {})",
addr
)
}
async fn on_observe_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>, value: f64);
async fn on_observe_bool(&mut self, addr: &Address, dist: &dyn Distribution<bool>, value: bool);
async fn on_observe_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>, value: u64);
async fn on_observe_usize(
&mut self,
addr: &Address,
dist: &dyn Distribution<usize>,
value: usize,
);
async fn on_observe_i64(&mut self, addr: &Address, _dist: &dyn Distribution<i64>, _value: i64) {
panic!(
"handler does not implement on_observe_i64 (i64 observe site at {})",
addr
)
}
async fn on_factor(&mut self, logw: f64);
async fn finish(self) -> Trace
where
Self: Sized;
}
#[derive(Clone, Copy, Debug, Default)]
pub struct FromSync<H>(pub H);
impl<H: Handler> AsyncHandler for FromSync<H> {
async fn on_sample_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>) -> f64 {
self.0.on_sample_f64(addr, dist)
}
async fn on_sample_bool(&mut self, addr: &Address, dist: &dyn Distribution<bool>) -> bool {
self.0.on_sample_bool(addr, dist)
}
async fn on_sample_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>) -> u64 {
self.0.on_sample_u64(addr, dist)
}
async fn on_sample_usize(&mut self, addr: &Address, dist: &dyn Distribution<usize>) -> usize {
self.0.on_sample_usize(addr, dist)
}
async fn on_sample_i64(&mut self, addr: &Address, dist: &dyn Distribution<i64>) -> i64 {
self.0.on_sample_i64(addr, dist)
}
async fn on_observe_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>, value: f64) {
self.0.on_observe_f64(addr, dist, value)
}
async fn on_observe_bool(
&mut self,
addr: &Address,
dist: &dyn Distribution<bool>,
value: bool,
) {
self.0.on_observe_bool(addr, dist, value)
}
async fn on_observe_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>, value: u64) {
self.0.on_observe_u64(addr, dist, value)
}
async fn on_observe_usize(
&mut self,
addr: &Address,
dist: &dyn Distribution<usize>,
value: usize,
) {
self.0.on_observe_usize(addr, dist, value)
}
async fn on_observe_i64(&mut self, addr: &Address, dist: &dyn Distribution<i64>, value: i64) {
self.0.on_observe_i64(addr, dist, value)
}
async fn on_factor(&mut self, logw: f64) {
self.0.on_factor(logw)
}
async fn finish(self) -> Trace {
self.0.finish()
}
}
pub async fn run_async<A, H: AsyncHandler>(mut h: H, m: Model<A>) -> (A, Trace) {
let mut m = m;
let a = loop {
m = match m {
Model::Pure(a) => break a,
Model::SampleF64 { addr, dist, k } => {
let x = h.on_sample_f64(&addr, &*dist).await;
k(x)
}
Model::SampleBool { addr, dist, k } => {
let x = h.on_sample_bool(&addr, &*dist).await;
k(x)
}
Model::SampleU64 { addr, dist, k } => {
let x = h.on_sample_u64(&addr, &*dist).await;
k(x)
}
Model::SampleUsize { addr, dist, k } => {
let x = h.on_sample_usize(&addr, &*dist).await;
k(x)
}
Model::SampleI64 { addr, dist, k } => {
let x = h.on_sample_i64(&addr, &*dist).await;
k(x)
}
Model::ObserveF64 {
addr,
dist,
value,
k,
} => {
h.on_observe_f64(&addr, &*dist, value).await;
k(())
}
Model::ObserveBool {
addr,
dist,
value,
k,
} => {
h.on_observe_bool(&addr, &*dist, value).await;
k(())
}
Model::ObserveU64 {
addr,
dist,
value,
k,
} => {
h.on_observe_u64(&addr, &*dist, value).await;
k(())
}
Model::ObserveUsize {
addr,
dist,
value,
k,
} => {
h.on_observe_usize(&addr, &*dist, value).await;
k(())
}
Model::ObserveI64 {
addr,
dist,
value,
k,
} => {
h.on_observe_i64(&addr, &*dist, value).await;
k(())
}
Model::Factor { logw, k } => {
h.on_factor(logw).await;
k(())
}
};
};
let t = h.finish().await;
(a, t)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::addr;
use crate::core::distribution::*;
use crate::core::model::{factor, observe, pure, sample, ModelExt};
use crate::runtime::handler::run;
use crate::runtime::interpreters::{PriorHandler, ReplayHandler, ScoreGivenTrace};
use rand::rngs::StdRng;
use rand::{RngCore, SeedableRng};
use std::future::Future;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
fn block_on<F: Future>(fut: F) -> F::Output {
struct ThreadWaker(std::thread::Thread);
impl Wake for ThreadWaker {
fn wake(self: Arc<Self>) {
self.0.unpark();
}
}
let waker = Waker::from(Arc::new(ThreadWaker(std::thread::current())));
let mut cx = Context::from_waker(&waker);
let mut fut = std::pin::pin!(fut);
loop {
match fut.as_mut().poll(&mut cx) {
Poll::Ready(out) => return out,
Poll::Pending => std::thread::park(),
}
}
}
struct YieldNow(bool);
impl Future for YieldNow {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
if self.0 {
Poll::Ready(())
} else {
self.0 = true;
cx.waker().wake_by_ref();
Poll::Pending
}
}
}
struct Yielding<H>(H);
impl<H: AsyncHandler> AsyncHandler for Yielding<H> {
async fn on_sample_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>) -> f64 {
YieldNow(false).await;
self.0.on_sample_f64(addr, dist).await
}
async fn on_sample_bool(&mut self, addr: &Address, dist: &dyn Distribution<bool>) -> bool {
YieldNow(false).await;
self.0.on_sample_bool(addr, dist).await
}
async fn on_sample_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>) -> u64 {
YieldNow(false).await;
self.0.on_sample_u64(addr, dist).await
}
async fn on_sample_usize(
&mut self,
addr: &Address,
dist: &dyn Distribution<usize>,
) -> usize {
YieldNow(false).await;
self.0.on_sample_usize(addr, dist).await
}
async fn on_sample_i64(&mut self, addr: &Address, dist: &dyn Distribution<i64>) -> i64 {
YieldNow(false).await;
self.0.on_sample_i64(addr, dist).await
}
async fn on_observe_f64(&mut self, addr: &Address, dist: &dyn Distribution<f64>, v: f64) {
YieldNow(false).await;
self.0.on_observe_f64(addr, dist, v).await
}
async fn on_observe_bool(
&mut self,
addr: &Address,
dist: &dyn Distribution<bool>,
v: bool,
) {
YieldNow(false).await;
self.0.on_observe_bool(addr, dist, v).await
}
async fn on_observe_u64(&mut self, addr: &Address, dist: &dyn Distribution<u64>, v: u64) {
YieldNow(false).await;
self.0.on_observe_u64(addr, dist, v).await
}
async fn on_observe_usize(
&mut self,
addr: &Address,
dist: &dyn Distribution<usize>,
v: usize,
) {
YieldNow(false).await;
self.0.on_observe_usize(addr, dist, v).await
}
async fn on_observe_i64(&mut self, addr: &Address, dist: &dyn Distribution<i64>, v: i64) {
YieldNow(false).await;
self.0.on_observe_i64(addr, dist, v).await
}
async fn on_factor(&mut self, logw: f64) {
YieldNow(false).await;
self.0.on_factor(logw).await
}
async fn finish(self) -> Trace {
YieldNow(false).await;
self.0.finish().await
}
}
type AllTypes = (f64, bool, u64, usize, i64);
fn all_effects_model() -> Model<AllTypes> {
crate::prob! {
let x <- sample(addr!("x"), Normal::new(0.0, 1.0).unwrap());
let b <- sample(addr!("b"), Bernoulli::new(0.3).unwrap());
let n <- sample(addr!("n"), Poisson::new(4.0).unwrap());
let c <- sample(addr!("c"), Categorical::new(vec![0.2, 0.5, 0.3]).unwrap());
let z <- sample(addr!("z"), DiscreteUniform::new(-3, 3).unwrap());
observe(addr!("obs_f64"), Normal::new(x, 0.5).unwrap(), 0.4);
observe(addr!("obs_bool"), Bernoulli::new(if b { 0.9 } else { 0.2 }).unwrap(), true);
observe(addr!("obs_u64"), Poisson::new(n as f64 + 1.0).unwrap(), 3u64);
observe(addr!("obs_usize"), Categorical::uniform(3).unwrap(), c);
observe(addr!("obs_i64"), DiscreteUniform::new(z - 1, z + 1).unwrap(), z);
factor(-0.5 * x * x);
pure((x, b, n, c, z))
}
}
fn assert_same_run(expected: &(AllTypes, Trace), actual: &(AllTypes, Trace)) {
let ((x, b, n, c, z), t) = expected;
let ((ax, ab, an, ac, az), at) = actual;
assert_eq!(x.to_bits(), ax.to_bits(), "value x");
assert_eq!((b, n, c, z), (ab, an, ac, az), "value");
assert_eq!(t.choices.len(), at.choices.len(), "number of choices");
for ((addr, choice), (a_addr, a_choice)) in t.choices.iter().zip(&at.choices) {
assert_eq!(addr, a_addr);
assert_eq!(choice.addr, a_choice.addr);
assert_eq!(choice.value, a_choice.value, "value at {addr}");
assert_eq!(
choice.logp.to_bits(),
a_choice.logp.to_bits(),
"logp at {addr}"
);
}
assert_eq!(t.log_prior.to_bits(), at.log_prior.to_bits(), "log_prior");
assert_eq!(
t.log_likelihood.to_bits(),
at.log_likelihood.to_bits(),
"log_likelihood"
);
assert_eq!(
t.log_factors.to_bits(),
at.log_factors.to_bits(),
"log_factors"
);
}
#[test]
fn from_sync_matches_run_for_builtin_handlers() {
let mut rngs = [0, 1, 2].map(|_| StdRng::seed_from_u64(62));
let [r_sync, r_async, r_yield] = &mut rngs;
let prior = |rng| PriorHandler {
rng,
trace: Trace::default(),
};
let expected = run(prior(r_sync), all_effects_model());
let from_sync = block_on(run_async(FromSync(prior(r_async)), all_effects_model()));
let yielding = block_on(run_async(
Yielding(FromSync(prior(r_yield))),
all_effects_model(),
));
assert_same_run(&expected, &from_sync);
assert_same_run(&expected, &yielding);
let next = rngs.map(|mut rng| rng.next_u64());
assert!(next[0] == next[1] && next[0] == next[2], "RNG consumption");
let prior_trace = expected.1;
assert_eq!(prior_trace.choices.len(), 5);
assert!(prior_trace.log_prior.is_finite());
assert!(prior_trace.log_likelihood.is_finite() && prior_trace.log_likelihood < 0.0);
assert!(prior_trace.log_factors <= 0.0);
let mut base = prior_trace.clone();
base.choices.remove(&addr!("n"));
let mut rngs = [0, 1, 2].map(|_| StdRng::seed_from_u64(7));
let [r_sync, r_async, r_yield] = &mut rngs;
let replay = |rng| ReplayHandler {
rng,
base: base.clone(),
trace: Trace::default(),
};
let expected = run(replay(r_sync), all_effects_model());
let from_sync = block_on(run_async(FromSync(replay(r_async)), all_effects_model()));
let yielding = block_on(run_async(
Yielding(FromSync(replay(r_yield))),
all_effects_model(),
));
assert_same_run(&expected, &from_sync);
assert_same_run(&expected, &yielding);
let next = rngs.map(|mut rng| rng.next_u64());
assert!(next[0] == next[1] && next[0] == next[2], "RNG consumption");
let score = || ScoreGivenTrace {
base: prior_trace.clone(),
trace: Trace::default(),
};
let expected = run(score(), all_effects_model());
let from_sync = block_on(run_async(FromSync(score()), all_effects_model()));
let yielding = block_on(run_async(Yielding(FromSync(score())), all_effects_model()));
assert_same_run(&expected, &from_sync);
assert_same_run(&expected, &yielding);
assert_eq!(
expected.1.total_log_weight().to_bits(),
prior_trace.total_log_weight().to_bits()
);
}
#[test]
fn run_async_is_stack_safe_for_deep_models() {
fn build(i: usize, n: usize, acc: f64) -> Model<f64> {
if i >= n {
pure(acc)
} else {
sample(addr!("x", i), Normal::new(0.0, 1.0).unwrap())
.bind(move |x| build(i + 1, n, acc + x))
}
}
let handle = std::thread::Builder::new()
.stack_size(512 * 1024)
.spawn(|| {
let n = 100_000;
let mut rng = StdRng::seed_from_u64(2024);
let (sum, trace) = block_on(run_async(
FromSync(PriorHandler {
rng: &mut rng,
trace: Trace::default(),
}),
build(0, n, 0.0),
));
assert!(sum.is_finite());
assert_eq!(trace.choices.len(), n);
assert!(trace.log_prior.is_finite());
let mut rng = StdRng::seed_from_u64(2024);
let (sum_yielding, trace_yielding) = block_on(run_async(
Yielding(FromSync(PriorHandler {
rng: &mut rng,
trace: Trace::default(),
})),
build(0, n, 0.0),
));
assert_eq!(sum_yielding.to_bits(), sum.to_bits());
assert_eq!(trace_yielding.choices.len(), n);
assert_eq!(
trace_yielding.log_prior.to_bits(),
trace.log_prior.to_bits()
);
})
.expect("spawn thread");
handle
.join()
.expect("deep model async interpretation overflowed the stack");
}
struct NoI64;
impl Handler for NoI64 {
fn on_sample_f64(&mut self, _: &Address, _: &dyn Distribution<f64>) -> f64 {
unreachable!()
}
fn on_sample_bool(&mut self, _: &Address, _: &dyn Distribution<bool>) -> bool {
unreachable!()
}
fn on_sample_u64(&mut self, _: &Address, _: &dyn Distribution<u64>) -> u64 {
unreachable!()
}
fn on_sample_usize(&mut self, _: &Address, _: &dyn Distribution<usize>) -> usize {
unreachable!()
}
fn on_observe_f64(&mut self, _: &Address, _: &dyn Distribution<f64>, _: f64) {
unreachable!()
}
fn on_observe_bool(&mut self, _: &Address, _: &dyn Distribution<bool>, _: bool) {
unreachable!()
}
fn on_observe_u64(&mut self, _: &Address, _: &dyn Distribution<u64>, _: u64) {
unreachable!()
}
fn on_observe_usize(&mut self, _: &Address, _: &dyn Distribution<usize>, _: usize) {
unreachable!()
}
fn on_factor(&mut self, _: f64) {
unreachable!()
}
fn finish(self) -> Trace {
unreachable!()
}
}
impl AsyncHandler for NoI64 {
async fn on_sample_f64(&mut self, _: &Address, _: &dyn Distribution<f64>) -> f64 {
unreachable!()
}
async fn on_sample_bool(&mut self, _: &Address, _: &dyn Distribution<bool>) -> bool {
unreachable!()
}
async fn on_sample_u64(&mut self, _: &Address, _: &dyn Distribution<u64>) -> u64 {
unreachable!()
}
async fn on_sample_usize(&mut self, _: &Address, _: &dyn Distribution<usize>) -> usize {
unreachable!()
}
async fn on_observe_f64(&mut self, _: &Address, _: &dyn Distribution<f64>, _: f64) {
unreachable!()
}
async fn on_observe_bool(&mut self, _: &Address, _: &dyn Distribution<bool>, _: bool) {
unreachable!()
}
async fn on_observe_u64(&mut self, _: &Address, _: &dyn Distribution<u64>, _: u64) {
unreachable!()
}
async fn on_observe_usize(&mut self, _: &Address, _: &dyn Distribution<usize>, _: usize) {
unreachable!()
}
async fn on_factor(&mut self, _: f64) {
unreachable!()
}
async fn finish(self) -> Trace {
unreachable!()
}
}
fn panic_message(f: impl FnOnce()) -> String {
let payload = catch_unwind(AssertUnwindSafe(f)).expect_err("expected a panic");
match payload.downcast::<String>() {
Ok(message) => *message,
Err(payload) => payload
.downcast_ref::<&str>()
.expect("panic payload is a string")
.to_string(),
}
}
#[test]
fn i64_defaults_panic_with_the_handler_message() {
let sample_i64 = || sample(addr!("z"), DiscreteUniform::new(-1, 1).unwrap());
let documented = "handler does not implement on_sample_i64 (i64 sample site at z)";
assert_eq!(
panic_message(|| drop(block_on(run_async(NoI64, sample_i64())))),
documented
);
assert_eq!(panic_message(|| drop(run(NoI64, sample_i64()))), documented);
assert_eq!(
panic_message(|| drop(block_on(run_async(FromSync(NoI64), sample_i64())))),
documented
);
let observe_i64 = || observe(addr!("w"), DiscreteUniform::new(-1, 1).unwrap(), 0);
let documented = "handler does not implement on_observe_i64 (i64 observe site at w)";
assert_eq!(
panic_message(|| drop(block_on(run_async(NoI64, observe_i64())))),
documented
);
assert_eq!(
panic_message(|| drop(run(NoI64, observe_i64()))),
documented
);
assert_eq!(
panic_message(|| drop(block_on(run_async(FromSync(NoI64), observe_i64())))),
documented
);
}
}