dope 0.2.3

Thin io_uring adaptor with "Manifolds"
Documentation
use std::time::{Duration, Instant};

#[test]
fn sleep_zero_returns_immediately() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let start = Instant::now();
        dope_runtime::time::sleep_with(ctx, Duration::ZERO)
            .await
            .unwrap();
        assert!(start.elapsed() < Duration::from_millis(10));
    });
}

#[test]
fn sleep_short_duration() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let start = Instant::now();
        dope_runtime::time::sleep_with(ctx, Duration::from_millis(50))
            .await
            .unwrap();
        let elapsed = start.elapsed();
        assert!(
            elapsed >= Duration::from_millis(40),
            "too fast: {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_millis(200),
            "too slow: {elapsed:?}"
        );
    });
}

#[test]
fn sleep_until_in_past() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let past = Instant::now() - Duration::from_secs(1);
        let start = Instant::now();
        dope_runtime::time::sleep_with(ctx, past.saturating_duration_since(Instant::now()))
            .await
            .unwrap();
        assert!(start.elapsed() < Duration::from_millis(10));
    });
}

#[test]
fn multiple_concurrent_sleeps() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let start = Instant::now();

        let (a, b, c) = futures_join3(
            dope_runtime::time::sleep_with(ctx, Duration::from_millis(30)),
            dope_runtime::time::sleep_with(ctx, Duration::from_millis(60)),
            dope_runtime::time::sleep_with(ctx, Duration::from_millis(90)),
        )
        .await;
        a.unwrap();
        b.unwrap();
        c.unwrap();

        let elapsed = start.elapsed();
        assert!(
            elapsed >= Duration::from_millis(80),
            "too fast: {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_millis(300),
            "too slow: {elapsed:?}"
        );
    });
}

#[test]
fn timeout_fires_before_slow_future() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let result = dope_runtime::time::timeout_with(ctx, Duration::from_millis(30), async {
            dope_runtime::time::sleep_with(ctx, Duration::from_secs(10))
                .await
                .unwrap();
            42
        })
        .await;
        assert!(result.is_err());
    });
}

#[test]
fn timeout_does_not_fire_for_fast_future() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        let result = dope_runtime::time::timeout_with(ctx, Duration::from_secs(10), async {
            dope_runtime::time::sleep_with(ctx, Duration::from_millis(10))
                .await
                .unwrap();
            42
        })
        .await;
        assert_eq!(result.unwrap(), 42);
    });
}

#[test]
fn cancelled_sleep_does_not_leak() {
    let mut rt = dope_runtime::runtime::Runtime::<
        dope_runtime::executor::LocalExecutor<
            dope_runtime::DefaultBackend,
            dope_core::profile::Throughput,
        >,
    >::new_local()
    .unwrap();
    let ctx = rt.ctx();
    rt.run_until(async move {
        for _ in 0..1000 {
            let fut = dope_runtime::time::sleep_with(ctx, Duration::from_secs(999));
            drop(fut);
        }
        dope_runtime::time::sleep_with(ctx, Duration::from_millis(10))
            .await
            .unwrap();
    });
}

async fn futures_join3<A, B, C>(a: A, b: B, c: C) -> (A::Output, B::Output, C::Output)
where
    A: std::future::Future,
    B: std::future::Future,
    C: std::future::Future,
{
    use std::future::Future;
    use std::pin::Pin;
    use std::task::{Context, Poll};

    struct Join3<A: Future, B: Future, C: Future> {
        a: Pin<Box<A>>,
        b: Pin<Box<B>>,
        c: Pin<Box<C>>,
        out_a: Option<A::Output>,
        out_b: Option<B::Output>,
        out_c: Option<C::Output>,
    }

    impl<A: Future, B: Future, C: Future> Future for Join3<A, B, C> {
        type Output = (A::Output, B::Output, C::Output);

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            let me = unsafe { self.get_unchecked_mut() };
            if me.out_a.is_none()
                && let Poll::Ready(v) = me.a.as_mut().poll(cx)
            {
                me.out_a = Some(v);
            }
            if me.out_b.is_none()
                && let Poll::Ready(v) = me.b.as_mut().poll(cx)
            {
                me.out_b = Some(v);
            }
            if me.out_c.is_none()
                && let Poll::Ready(v) = me.c.as_mut().poll(cx)
            {
                me.out_c = Some(v);
            }
            if me.out_a.is_some() && me.out_b.is_some() && me.out_c.is_some() {
                Poll::Ready((
                    me.out_a.take().unwrap(),
                    me.out_b.take().unwrap(),
                    me.out_c.take().unwrap(),
                ))
            } else {
                Poll::Pending
            }
        }
    }

    Join3 {
        a: Box::pin(a),
        b: Box::pin(b),
        c: Box::pin(c),
        out_a: None,
        out_b: None,
        out_c: None,
    }
    .await
}