pipecircuit 0.0.21

#![cfg(feature = "storage")]
use core::num::{NonZeroU32, NonZeroU64};
use pipecircuit::{
    ContinuationAdmissionOutcome, ContinuationArmOutcome, ContinuationCancellationOutcome,
    ContinuationLedger, ContinuationResolutionOutcome, Owed, Pipe, TimeoutAfter, TimerOut,
    storage::warp_table::Capacity,
};
#[global_allocator]
static ALLOCATOR: dhat::Alloc = dhat::Alloc;
struct Worker;
impl Pipe for Worker {
    type In<'a> = Box<[u8; 1024]>;
    type Out<'a> = Box<[u8; 1024]>;
}
#[test]
fn deferred_round_trips_reuse_owned_storage_without_allocating() -> Result<(), &'static str> {
    let mut ledger =
        ContinuationLedger::try_new(Capacity::new(NonZeroU32::MIN)).map_err(|_| "ledger")?;
    let mut payload = Box::new([0_u8; 1024]);
    let pointer = core::ptr::from_ref(payload.as_ref());
    let _profiler = dhat::Profiler::builder().testing().build();
    for _ in 0..256 {
        let ContinuationAdmissionOutcome::Admitted { admission } = ledger.admit::<Worker>(
            7_u8,
            Owed::new(payload, TimeoutAfter::seconds(NonZeroU64::MIN)),
        ) else {
            return Err("admission");
        };
        let armed = TimerOut::Armed {
            token: admission.timer_token(),
        }
        .into_arm_outcome()
        .map_err(|_| "arm output")?;
        let ContinuationArmOutcome::Armed { work } = admission
            .settle_timer_arm(armed)
            .map_err(|_| "arm mismatch")?
        else {
            return Err("arm refused");
        };
        let (input, reply) = work.begin();
        let completion = reply.fulfill(input);
        let cancelled = TimerOut::Cancelled {
            token: completion.timer_token(),
        }
        .into_cancellation_outcome()
        .map_err(|_| "cancel output")?;
        let ContinuationCancellationOutcome::Cancelled { completion } = completion
            .settle_timer_cancellation(cancelled)
            .map_err(|_| "cancel mismatch")?
        else {
            return Err("stale cancellation");
        };
        let ContinuationResolutionOutcome::Resolved {
            return_to: 7,
            result: Ok(output),
        } = ledger.resolve(completion)
        else {
            return Err("resolution");
        };
        payload = output;
        assert_eq!(core::ptr::from_ref(payload.as_ref()), pointer);
    }
    let stats = dhat::HeapStats::get();
    assert_eq!(stats.total_blocks, 0);
    assert_eq!(stats.total_bytes, 0);
    assert!(ledger.is_empty());
    Ok(())
}