some_executor 0.7.2

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

/*!
This executor is in use when no other executors are registered.

It is intentionally the simplest idea possible, but it ensures a compliant executor is always available.

> Cut my tasks into pieces, this is my last resort!
> Async handling, no tokio, don't give a fuck if performance is bleeding
> this is my last resort

*/

use crate::observer::{FinishedObservation, Observer, ObserverNotified};
use crate::task::Task;
use crate::{DynExecutor, SomeExecutor};
use std::any::Any;
use std::convert::Infallible;
use std::future::Future;
use std::pin::Pin;
#[cfg(not(target_arch = "wasm32"))]
use std::pin::pin;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::Arc;
#[cfg(not(target_arch = "wasm32"))]
use std::sync::atomic::{AtomicU8, Ordering};
#[cfg(not(target_arch = "wasm32"))]
use std::task::{Context, Poll, RawWaker, RawWakerVTable};
#[cfg(not(target_arch = "wasm32"))]
use wasm_lite_std::Mutex;
#[cfg(not(target_arch = "wasm32"))]
use wasm_lite_std::condvar::Condvar;

#[cfg(not(target_arch = "wasm32"))]
use std::thread;
#[cfg(target_arch = "wasm32")]
use wasm_lite_std as thread;

#[derive(Debug)]
pub(crate) struct LastResortExecutor;

impl LastResortExecutor {
    pub fn new() -> Self {
        LastResortExecutor
    }
}

#[cfg(not(target_arch = "wasm32"))]
const SLEEPING: u8 = 0;
#[cfg(not(target_arch = "wasm32"))]
const LISTENING: u8 = 1;
#[cfg(not(target_arch = "wasm32"))]
const WAKEPLS: u8 = 2;

#[cfg(not(target_arch = "wasm32"))]
struct Shared {
    condvar: Condvar,
    mutex: Mutex<bool>,
    //have to be careful about deadlocks here,
    inline_notify: AtomicU8,
}
#[cfg(not(target_arch = "wasm32"))]
struct Waker {
    shared: Arc<Shared>,
}

#[cfg(not(target_arch = "wasm32"))]
const WAKER_VTABLE: RawWakerVTable = RawWakerVTable::new(
    |data| {
        let waker = unsafe { Arc::from_raw(data as *const Waker) };
        let w2 = waker.clone();
        std::mem::forget(waker);
        RawWaker::new(Arc::into_raw(w2) as *const (), &WAKER_VTABLE)
    },
    |data| {
        let waker = unsafe { Arc::from_raw(data as *const Waker) };
        let old = waker.shared.inline_notify.swap(WAKEPLS, Ordering::AcqRel);
        if old == SLEEPING {
            //Acquire mutex before notify: otherwise the notify can fire in the window
            //between the executor observing SLEEPING and actually parking on the
            //condvar, and the wakeup is lost forever.
            let _guard = waker.shared.mutex.lock_sync();
            waker.shared.condvar.notify_one();
        }
        drop(waker);
    },
    |data| {
        let waker = unsafe { Arc::from_raw(data as *const Waker) };
        let old = waker.shared.inline_notify.swap(WAKEPLS, Ordering::AcqRel);
        if old == SLEEPING {
            //Acquire mutex before notify: otherwise the notify can fire in the window
            //between the executor observing SLEEPING and actually parking on the
            //condvar, and the wakeup is lost forever.
            let _guard = waker.shared.mutex.lock_sync();
            waker.shared.condvar.notify_one();
        }
        std::mem::forget(waker);
    },
    |data| {
        let waker = unsafe { Arc::from_raw(data as *const Waker) };
        drop(waker)
    },
);
#[cfg(not(target_arch = "wasm32"))]
impl Waker {
    fn into_core_waker(self) -> core::task::Waker {
        let data = Arc::into_raw(Arc::new(self));
        unsafe { core::task::Waker::from_raw(RawWaker::new(data as *const (), &WAKER_VTABLE)) }
    }
}

impl LastResortExecutor {
    fn print_warning_or_panic() {
        crate::warn_or_panic(
            "some_executor::LastResortExecutor is in use.  This is not intended for production code; investigate ways to use a production-quality executor. Set SOME_EXECUTOR_BUILTIN_SHOULD_PANIC=1 to panic instead.",
        );
    }

    #[cfg(not(target_arch = "wasm32"))]
    fn spawn<F: Future + Send + 'static>(f: F, poll_after: crate::sys::Instant) {
        thread::Builder::new()
            .name("some_executor::LastResortExecutor".to_string())
            .spawn(move || {
                //respect poll_after: conforming executors must not poll before this
                //time (common_poll asserts it)
                let now = crate::sys::Instant::now();
                if poll_after > now {
                    thread::sleep(poll_after.duration_since(now));
                }
                let shared = Arc::new(Shared {
                    condvar: Condvar::new(),
                    mutex: Mutex::new(false),
                    inline_notify: AtomicU8::new(SLEEPING),
                });
                let waker = Waker {
                    shared: shared.clone(),
                }
                .into_core_waker();
                let mut c = Context::from_waker(&waker);
                let mut pin = pin!(f);
                loop {
                    let mut _guard = shared.mutex.lock_sync();
                    //eagerly poll
                    shared.inline_notify.store(LISTENING, Ordering::Relaxed);
                    let r = pin.as_mut().poll(&mut c);
                    match r {
                        Poll::Ready(..) => {
                            return;
                        }
                        Poll::Pending => {
                            let old = shared.inline_notify.swap(SLEEPING, Ordering::AcqRel);
                            if old == WAKEPLS {
                                //release lock anyway
                                drop(_guard);
                                continue; //poll eagerly
                            } else {
                                _guard = shared.condvar.wait_sync_while(_guard, |_| {
                                    shared.inline_notify.load(Ordering::Acquire) != WAKEPLS
                                });
                            }
                        }
                    }
                }
            })
            .unwrap();
    }

    #[cfg(target_arch = "wasm32")]
    fn spawn<F: Future + Send + 'static>(f: F, poll_after: crate::sys::Instant) {
        thread::Builder::new()
            .name("some_executor::LastResortExecutor".to_string())
            .spawn(move || {
                thread::task_begin();
                wasm_lite_std::spawn_local(async move {
                    struct TaskFinishedOnDrop;
                    impl Drop for TaskFinishedOnDrop {
                        fn drop(&mut self) {
                            thread::task_finished();
                        }
                    }
                    let _task_finished_on_drop = TaskFinishedOnDrop;
                    //respect poll_after: conforming executors must not poll before
                    //this time (common_poll asserts it)
                    crate::sys::wait_until(poll_after).await;
                    f.await;
                });
            })
            .unwrap();
    }
}

impl SomeExecutor for LastResortExecutor {
    type ExecutorNotifier = Infallible;

    fn spawn<F: Future + Send + 'static, Notifier: ObserverNotified<F::Output> + Send>(
        &mut self,
        task: Task<F, Notifier>,
    ) -> impl Observer<Value = F::Output>
    where
        Self: Sized,
        F::Output: Send + Unpin,
    {
        Self::print_warning_or_panic();
        let (s, o) = task.spawn(self);
        let poll_after = s.poll_after();
        Self::spawn(s, poll_after);
        o
    }

    async fn spawn_async<F: Future + Send + 'static, Notifier: ObserverNotified<F::Output> + Send>(
        &mut self,
        task: Task<F, Notifier>,
    ) -> impl Observer<Value = F::Output>
    where
        Self: Sized,
        F::Output: Send + Unpin,
    {
        Self::print_warning_or_panic();
        let (s, o) = task.spawn(self);
        let poll_after = s.poll_after();
        Self::spawn(s, poll_after);
        o
    }

    fn spawn_objsafe(
        &mut self,
        task: Task<
            Pin<Box<dyn Future<Output = Box<dyn Any + 'static + Send>> + 'static + Send>>,
            Box<dyn ObserverNotified<dyn Any + Send> + Send>,
        >,
    ) -> Box<
        dyn Observer<Value = Box<dyn Any + Send>, Output = FinishedObservation<Box<dyn Any + Send>>>
            + Send,
    > {
        Self::print_warning_or_panic();
        let (s, o) = task.spawn_objsafe(self);
        let poll_after = s.poll_after();
        Self::spawn(s, poll_after);
        Box::new(o)
    }

    fn spawn_objsafe_async<'s>(
        &'s mut self,
        task: Task<
            Pin<Box<dyn Future<Output = Box<dyn Any + 'static + Send>> + 'static + Send>>,
            Box<dyn ObserverNotified<dyn Any + Send> + Send>,
        >,
    ) -> Box<
        dyn Future<
                Output = Box<
                    dyn Observer<
                            Value = Box<dyn Any + Send>,
                            Output = FinishedObservation<Box<dyn Any + Send>>,
                        > + Send,
                >,
            > + 's,
    > {
        Self::print_warning_or_panic();
        #[allow(clippy::async_yields_async)]
        Box::new(async {
            let (s, o) = task.spawn_objsafe(self);
            let poll_after = s.poll_after();
            Self::spawn(s, poll_after);
            Box::new(o)
                as Box<
                    dyn Observer<
                            Value = Box<dyn Any + Send>,
                            Output = FinishedObservation<Box<dyn Any + Send>>,
                        > + Send,
                >
        })
    }

    fn clone_box(&self) -> Box<DynExecutor> {
        Box::new(Self)
    }

    fn executor_notifier(&mut self) -> Option<Self::ExecutorNotifier> {
        None
    }
}