some_global_executor 0.1.6

// SPDX-License-Identifier: MIT OR Apache-2.0

//! The crate's own unit tests.
//!
//! Split out of `lib.rs` so the module a reader opens to learn what this
//! crate does is the public surface, not eight hundred lines of assertions.

use some_executor::SomeExecutor;
use some_executor::observer::Observation;
use some_executor::observer::Observer;
use some_executor::task::Configuration;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
#[cfg(not(target_arch = "wasm32"))]
use std::thread;
#[cfg(target_arch = "wasm32")]
use wasm_lite_std as thread;

#[wasm_lite::wasm_lite_test]
fn new() {
    let e = super::Executor::new("test".to_string(), 4);
    e.drain();
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
#[should_panic(expected = "at least one worker thread")]
fn rejects_zero_workers() {
    let _ = super::Executor::new("zero".to_string(), 0);
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
#[should_panic(expected = "at least one worker thread")]
fn rejects_resize_to_zero() {
    let mut executor = super::Executor::new("resize-zero".to_string(), 1);
    executor.resize(0);
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn executor_name_may_contain_nul() {
    let name = "untrusted\0executor";
    let executor = super::Executor::new(name.to_string(), 1);

    assert_eq!(executor.name(), name);
    executor.drain();
}

#[wasm_lite::wasm_lite_test]
async fn respects_poll_after() {
    let deadline = some_executor::Instant::now() + std::time::Duration::from_millis(30);
    let configuration = some_executor::task::ConfigurationBuilder::new()
        .poll_after(deadline)
        .build();
    let mut executor = super::Executor::new("poll-after".to_string(), 1);
    let task = some_executor::task::Task::without_notifications(
        "delayed".to_string(),
        configuration,
        async { 42 },
    );
    let observer = executor.spawn(task);

    executor.drain_async().await;

    assert!(some_executor::Instant::now() >= deadline);
    assert_eq!(observer.observe(), Observation::Ready(42));
}

#[wasm_lite::wasm_lite_test]
async fn drain_waits_for_delayed_static_tasks() {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    let completed = Arc::new(AtomicBool::new(false));
    let completed_in_task = completed.clone();
    let deadline = some_executor::Instant::now() + std::time::Duration::from_millis(30);
    let mut executor = super::Executor::new("static-poll-after".to_string(), 1);
    let outer = some_executor::task::Task::without_notifications(
        "spawn delayed static".to_string(),
        Configuration::default(),
        async move {
            let configuration = some_executor::task::ConfigurationBuilder::new()
                .poll_after(deadline)
                .build();
            let inner = some_executor::task::Task::without_notifications(
                "delayed static".to_string(),
                configuration,
                async move {
                    completed_in_task.store(true, Ordering::Release);
                },
            );
            inner.spawn_static_current();
        },
    );
    executor.spawn(outer).detach();

    executor.drain_async().await;

    assert!(some_executor::Instant::now() >= deadline);
    assert!(completed.load(Ordering::Acquire));
}

/// A static task's observer must deliver the return value.
///
/// Regression test for the `into_future()` bug: `spawn_static` used to spawn
/// `spawned.into_future()`, which yields the bare inner future and drops the
/// wrapper around it -- including the `ObserverSender`, whose `Drop` sees the
/// observation still pending and marks it `Cancelled`. The future ran to
/// completion the whole time, so the failure was invisible to every existing
/// static test: they all `detach()` the observer and signal through an
/// `AtomicBool` side channel, and none of them ever looked at an observed
/// value. This one does.
///
/// It also covers the second half of the same bug: polling through the
/// wrapper is what installs the task-locals, so `TASK_ID` and `TASK_LABEL`
/// were empty inside a static task and are checked here.
#[wasm_lite::wasm_lite_test]
async fn static_observer_delivers_the_return_value() {
    use some_executor::observer::FinishedObservation;
    use some_executor::task::{TASK_ID, TASK_LABEL};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    let ready = Arc::new(AtomicBool::new(false));
    let cancelled = Arc::new(AtomicBool::new(false));
    let ready_in_task = ready.clone();
    let cancelled_in_task = cancelled.clone();

    let mut executor = super::Executor::new("static-observer".to_string(), 1);
    let outer = some_executor::task::Task::without_notifications(
        "outer".to_string(),
        Configuration::default(),
        async move {
            // A static observer is !Send, so it can only be awaited from a
            // !Send task; hop onto the thread's static executor first.
            let checker = some_executor::task::Task::without_notifications(
                "checker".to_string(),
                Configuration::default(),
                async move {
                    let inner = some_executor::task::Task::without_notifications(
                        "inner-static".to_string(),
                        Configuration::default(),
                        async {
                            let id = TASK_ID.with(|id| id.copied());
                            let label = TASK_LABEL.with(|label| label.cloned());
                            (42i32, id.is_some(), label)
                        },
                    );
                    let observer = some_executor::thread_executor::thread_static_executor(|e| {
                        e.clone_box()
                            .spawn_static_objsafe(inner.into_objsafe_static())
                    });
                    match observer.await {
                        FinishedObservation::Ready(value) => {
                            let (answer, has_id, label) = *value
                                .downcast::<(i32, bool, Option<String>)>()
                                .expect("static observer delivered the wrong type");
                            assert_eq!(answer, 42, "the task's return value was corrupted");
                            assert!(has_id, "TASK_ID was not installed for a static task");
                            assert_eq!(
                                label.as_deref(),
                                Some("inner-static"),
                                "TASK_LABEL was not installed for a static task"
                            );
                            ready_in_task.store(true, Ordering::Release);
                        }
                        // Recorded rather than panicked: a panic inside a
                        // static task is isolated by the worker, so it would
                        // not fail this test.
                        FinishedObservation::Cancelled => {
                            cancelled_in_task.store(true, Ordering::Release);
                        }
                    }
                },
            );
            checker.spawn_static_current();
        },
    );
    executor.spawn(outer).detach();

    executor.drain_async().await;

    assert!(
        !cancelled.load(Ordering::Acquire),
        "static observer reported Cancelled; its ObserverSender was dropped before the task finished"
    );
    assert!(
        ready.load(Ordering::Acquire),
        "the checking task never observed a result"
    );
}

#[wasm_lite::wasm_lite_test]
async fn static_task_panic_does_not_kill_worker() {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    let survived = Arc::new(AtomicBool::new(false));
    let survived_in_task = survived.clone();
    let mut executor = super::Executor::new("static-panic-isolation".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "spawn static tasks".to_string(),
            Configuration::default(),
            async move {
                some_executor::task::Task::without_notifications(
                    "panic".to_string(),
                    Configuration::default(),
                    async { panic!("intentional static task panic") },
                )
                .spawn_static_current();
                some_executor::task::Task::without_notifications(
                    "survivor".to_string(),
                    Configuration::default(),
                    async move {
                        survived_in_task.store(true, Ordering::Release);
                    },
                )
                .spawn_static_current();
            },
        ))
        .detach();

    let deadline = some_executor::Instant::now() + std::time::Duration::from_secs(1);
    while !survived.load(Ordering::Acquire) && some_executor::Instant::now() < deadline {
        wasm_lite_std::sleep_async(std::time::Duration::from_millis(1)).await;
    }

    assert!(
        survived.load(Ordering::Acquire),
        "the worker died before polling the next static task"
    );
    executor.drain_async().await;
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn task_panic_does_not_kill_worker_or_hang_drain() {
    let mut executor = super::Executor::new("panic-isolation".to_string(), 1);
    let panicking = some_executor::task::Task::without_notifications(
        "panic".to_string(),
        Configuration::default(),
        async { panic!("intentional task panic") },
    );
    executor.spawn(panicking).detach();
    let survivor = some_executor::task::Task::without_notifications(
        "survivor".to_string(),
        Configuration::default(),
        async { 42 },
    );
    let observer = executor.spawn(survivor);

    executor.drain();

    assert_eq!(observer.observe(), Observation::Ready(42));
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn task_destructor_panic_does_not_kill_worker() {
    struct PanicsOnDrop;

    impl Future for PanicsOnDrop {
        type Output = ();

        fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
            Poll::Ready(())
        }
    }

    impl Drop for PanicsOnDrop {
        fn drop(&mut self) {
            panic!("intentional task destructor panic");
        }
    }

    let mut executor = super::Executor::new("drop-panic-isolation".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "drop panic".to_string(),
            Configuration::default(),
            PanicsOnDrop,
        ))
        .detach();
    let observer = executor.spawn(some_executor::task::Task::without_notifications(
        "survivor".to_string(),
        Configuration::default(),
        async { 42 },
    ));

    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
    loop {
        match observer.observe() {
            Observation::Ready(value) => {
                assert_eq!(value, 42);
                break;
            }
            Observation::Pending if std::time::Instant::now() < deadline => {
                std::thread::yield_now();
            }
            observation => panic!("survivor did not run after destructor panic: {observation:?}"),
        }
    }
    executor.drain();
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn static_task_destructor_panic_does_not_kill_worker() {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    struct PanicsOnDrop;

    impl Future for PanicsOnDrop {
        type Output = ();

        fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
            Poll::Ready(())
        }
    }

    impl Drop for PanicsOnDrop {
        fn drop(&mut self) {
            panic!("intentional static task destructor panic");
        }
    }

    let survived = Arc::new(AtomicBool::new(false));
    let survived_in_task = survived.clone();
    let mut executor = super::Executor::new("static-drop-panic-isolation".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "spawn static tasks".to_string(),
            Configuration::default(),
            async move {
                some_executor::task::Task::without_notifications(
                    "drop panic".to_string(),
                    Configuration::default(),
                    PanicsOnDrop,
                )
                .spawn_static_current();
                some_executor::task::Task::without_notifications(
                    "survivor".to_string(),
                    Configuration::default(),
                    async move {
                        survived_in_task.store(true, Ordering::Release);
                    },
                )
                .spawn_static_current();
            },
        ))
        .detach();

    executor.drain();

    assert!(
        survived.load(Ordering::Acquire),
        "the worker died before polling the next static task"
    );
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn wakes_all_concurrent_drainers() {
    use std::sync::mpsc;
    use std::time::Duration;

    let deadline = some_executor::Instant::now() + Duration::from_millis(30);
    let configuration = some_executor::task::ConfigurationBuilder::new()
        .poll_after(deadline)
        .build();
    let mut executor = super::Executor::new("multiple-drainers".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "delayed".to_string(),
            configuration,
            async {},
        ))
        .detach();

    let (finished, completion) = mpsc::channel();
    let first_drain = executor.clone().drain_async();
    let first_finished = finished.clone();
    let first = std::thread::spawn(move || {
        wasm_lite_std::block_on(first_drain);
        first_finished.send(()).unwrap();
    });
    let second_drain = executor.clone().drain_async();
    let second = std::thread::spawn(move || {
        wasm_lite_std::block_on(second_drain);
        finished.send(()).unwrap();
    });

    completion.recv_timeout(Duration::from_secs(1)).unwrap();
    completion.recv_timeout(Duration::from_secs(1)).unwrap();
    first.join().unwrap();
    second.join().unwrap();
    executor.drain();
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn dropping_drain_future_unregisters_its_waker() {
    use std::sync::Arc;
    use std::task::{Wake, Waker};

    struct CountingWake;
    #[allow(clippy::manual_noop_waker)]
    impl Wake for CountingWake {
        fn wake(self: Arc<Self>) {}
    }

    let (finish, pending) = r#continue::continuation();
    let mut executor = super::Executor::new("cancelled-drain".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "pending".to_string(),
            Configuration::default(),
            async move {
                pending.await;
            },
        ))
        .detach();

    let wake = Arc::new(CountingWake);
    let waker = Waker::from(wake.clone());
    let mut context = Context::from_waker(&waker);
    let mut drain = Box::pin(executor.clone().drain_async());
    assert!(drain.as_mut().poll(&mut context).is_pending());
    assert_eq!(Arc::strong_count(&wake), 3);

    drop(drain);
    assert_eq!(Arc::strong_count(&wake), 2);
    drop(waker);
    assert_eq!(Arc::strong_count(&wake), 1);

    finish.send(());
    executor.drain();
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn panicking_drain_waker_does_not_kill_worker() {
    use std::sync::Arc;
    use std::task::{Wake, Waker};

    struct PanickingWake;
    impl Wake for PanickingWake {
        fn wake(self: Arc<Self>) {
            panic!("intentional drain waker panic");
        }
    }

    let (finish, pending) = r#continue::continuation();
    let mut executor = super::Executor::new("drain-waker-panic".to_string(), 1);
    executor
        .spawn(some_executor::task::Task::without_notifications(
            "pending".to_string(),
            Configuration::default(),
            async move {
                pending.await;
            },
        ))
        .detach();

    let waker = Waker::from(Arc::new(PanickingWake));
    let mut context = Context::from_waker(&waker);
    let mut drain = Box::pin(executor.clone().drain_async());
    assert!(drain.as_mut().poll(&mut context).is_pending());
    finish.send(());
    std::thread::sleep(std::time::Duration::from_millis(20));

    let observer = executor.spawn(some_executor::task::Task::without_notifications(
        "survivor".to_string(),
        Configuration::default(),
        async { 42 },
    ));
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(1);
    loop {
        match observer.observe() {
            Observation::Ready(value) => {
                assert_eq!(value, 42);
                break;
            }
            Observation::Pending if std::time::Instant::now() < deadline => {
                std::thread::yield_now();
            }
            observation => panic!("survivor did not run after waker panic: {observation:?}"),
        }
    }
    drop(drain);
    executor.drain();
}

#[wasm_lite::wasm_lite_test]
async fn spawn() {
    let mut e = super::Executor::new("test".to_string(), 1);
    let (sender, fut) = r#continue::continuation();
    let t = some_executor::task::Task::without_notifications(
        "test spawn".to_string(),
        Configuration::default(),
        async move {
            sender.send(1);
        },
    );
    let _observer = e.spawn(t);
    let r = fut.await;
    assert_eq!(r, 1);
}

#[wasm_lite::wasm_lite_test]
async fn poll_count() {
    struct F(u32);
    impl Future for F {
        type Output = ();

        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            logwise::log!("poll_count is polling against {}", self.0);
            if self.0 == 0 {
                Poll::Ready(())
            } else {
                self.get_mut().0 -= 1;
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        }
    }
    let f = F(3);
    let mut e = super::Executor::new("poll_count".to_string(), 4);

    let task = some_executor::task::Task::without_notifications(
        "poll_count".to_string(),
        Configuration::default(),
        f,
    );

    let observer = e.spawn(task);
    let mut tries = 0;
    loop {
        let o = observer.observe();
        match o {
            Observation::Done => {
                panic!("done");
            }
            Observation::Ready(()) => break,
            Observation::Cancelled => {
                panic!("cancelled");
            }
            Observation::Pending => {
                tries += 1;
                if tries > 10000 {
                    panic!("too many tries");
                }
                wasm_lite_std::sleep_async(std::time::Duration::from_millis(1)).await;
            }
        }
    }
}

#[wasm_lite::wasm_lite_test]
async fn poll_outline() {
    struct F(u32);
    impl Future for F {
        type Output = ();

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
            if self.0 == 0 {
                Poll::Ready(())
            } else {
                let waker = cx.waker().clone();
                self.0 -= 1;
                thread::spawn(move || {
                    thread::sleep(std::time::Duration::from_millis(10));
                    waker.wake();
                });
                Poll::Pending
            }
        }
    }
    let f = F(10);
    let mut e = super::Executor::new("poll_count".to_string(), 4);
    let task = some_executor::task::Task::without_notifications(
        "poll_count".to_string(),
        Configuration::default(),
        f,
    );
    let observer = e.spawn(task);
    e.drain_async().await;
    assert_eq!(observer.observe(), Observation::Ready(()));
}