some_executor 0.7.2

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

/*!
Drives a future to completion from synchronous code.

This module provides the one-way door from sync into async: [`block_on`] parks the
calling thread until a future resolves, and hands back its output. It is the primitive
that lets `fn main`, a test, a CLI tool, or an FFI callback cross into async exactly
once and then stay executor-agnostic from there.

# Which `block_on` do I want?

There are two, and the difference matters:

- [`block_on`] (this function) polls the future **in place, on the calling thread**,
  parking between polls. It asks nothing of any executor. Use it when some *other*
  thread is making progress on whatever the future is waiting for.
- [`SomeExecutor::block_on`](crate::SomeExecutor::block_on) asks the executor to do it.
  Two kinds of backend must override
  [`SomeExecutor::block_on_objsafe`](crate::SomeExecutor::block_on_objsafe) rather than
  inherit the default: one that runs tasks on the calling thread, because parking the
  caller would stop the very loop the future depends on; and one whose resources need
  ambient thread-local context, because the default polls on a thread that has not
  entered it. tokio is the second kind even though its workers run elsewhere.

The second is implemented in terms of the first by default, which is correct for any
executor whose tasks make progress on other threads.

# When blocking is not possible

`block_on` works whenever the blocking thread can keep driving every scheduler the
future depends on. That covers the two ordinary cases — the executor runs on other
threads (park here, they drive), or the executor owns its own scheduling loop and runs
it here instead of parking.

It does not cover a future whose wakeups come from a scheduler the blocking thread
cannot drive without unwinding. The browser main thread is exactly that case: timers,
promises, `fetch`, and worker messages are all delivered by the JavaScript event loop,
and the event loop only turns when the stack unwinds. Blocking there starves the
mechanism that would end the block, so `block_on` panics on the wasm32 main thread
rather than hanging the tab. Use [`ExecutorMain`](crate::ExecutorMain) for a portable
entry point, or call `block_on` from a worker, where it works normally.

# Examples

```
use some_executor::block_on;

# // A doctest runs on the browser main thread, which cannot block; on wasm32 this
# // runs the body on a worker, where it can.  Natively it just calls the closure.
# wasm_lite_std::worker_doctest!(|| {
let answer = block_on(async { 6 * 7 });
assert_eq!(answer, 42);
# });
```

Futures may borrow from the stack — unlike spawning, `block_on` imposes no `Send` or
`'static` bound, because the future never leaves the calling thread:

```
use some_executor::block_on;

# wasm_lite_std::worker_doctest!(|| {
let name = String::from("world");
let greeting = block_on(async { format!("hello {}", name.as_str()) });
assert_eq!(greeting, "hello world");
// `name` was borrowed, not moved onto an executor.
assert_eq!(name, "world");
# });
```

# Stall detection

The most common way to misuse this is to let a current-thread executor inherit the
default [`SomeExecutor::block_on`](crate::SomeExecutor::block_on): the caller parks,
the executor stops, and nothing ever wakes. That failure is silent and looks like a
hang, so `block_on` watches for it. If the calling thread waits for a while without a
single wakeup, it prints a diagnostic naming the likely cause.

Set `SOME_EXECUTOR_BUILTIN_SHOULD_PANIC=1` to panic on that diagnostic instead of
printing it — the same switch the builtin executors use. Set
`SOME_EXECUTOR_BLOCK_ON_STALL_SECS` to change the threshold (default 10), or to `0` to
disable the check.
*/

use crate::sys::Instant;
use std::future::Future;
use std::time::Duration;

/// Default number of seconds without a wakeup before [`block_on`] reports a stall.
const DEFAULT_STALL_SECS: u64 = 10;

const STALL_MESSAGE: &str = "some_executor::block_on has waited a long time without a single wakeup, which usually \
means it can never be woken. Two things cause that. (1) The executor runs tasks on the calling thread -- a \
current-thread or local executor -- and inherited the default SomeExecutor::block_on, which parks the caller and \
so stops the very loop the future is waiting on; such an executor must override SomeExecutor::block_on_objsafe to \
run its own polling loop instead. (2) block_on was called from inside a task already running on this executor; \
re-entrancy is not supported. If instead this future is legitimately idle, set SOME_EXECUTOR_BLOCK_ON_STALL_SECS \
to raise the threshold (default 10) or to 0 to disable this check. Set SOME_EXECUTOR_BUILTIN_SHOULD_PANIC=1 to \
panic instead of printing this.";

#[cfg(target_arch = "wasm32")]
const MAIN_THREAD_MESSAGE: &str = "some_executor::block_on was called on the wasm32 main thread, which cannot block. \
The JavaScript event loop delivers every wakeup -- timers, promises, fetch, worker messages -- and it only runs \
once this thread's stack unwinds, so blocking here starves the wakeup that would end the block. Use \
some_executor::ExecutorMain for a portable entry point, or call block_on from a worker.";

/// Runs a future to completion on the calling thread, returning its output.
///
/// The future is polled in place; the thread parks between polls and wakes when the
/// future's waker fires. No `Send` or `'static` bound is required, because the future
/// never leaves this thread.
///
/// This function knows nothing about executors. Tasks spawned by the future run
/// wherever the ambient executor puts them; `block_on` only waits. See the
/// [module documentation](self) for when that is and is not sufficient, and for the
/// stall diagnostic that fires when it is not.
///
/// # Panics
///
/// On the wasm32 main thread, which cannot block. See the [module documentation](self).
///
/// Also panics if the stall diagnostic fires and
/// `SOME_EXECUTOR_BUILTIN_SHOULD_PANIC=1` is set.
///
/// # Examples
///
/// ```
/// # // Runs on a worker on wasm32, where blocking is legal; see Panics above.
/// # wasm_lite_std::worker_doctest!(|| {
/// let value = some_executor::block_on(async { 1 + 1 });
/// assert_eq!(value, 2);
/// # });
/// ```
pub fn block_on<F: Future>(future: F) -> F::Output {
    imp::block_on(future)
}

/// Runs `future` to completion with `executor` installed as the thread's executor.
///
/// This is the default body of [`SomeExecutor::block_on_objsafe`](crate::SomeExecutor::block_on_objsafe),
/// factored out so the install is scoped: code inside the block sees `executor` from
/// [`current_executor`](crate::current_executor::current_executor), and whatever
/// executor was there before is put back on the way out, panic or not.
pub(crate) fn block_on_installed<F: Future>(
    executor: Box<crate::DynExecutor>,
    future: F,
) -> F::Output {
    let _guard = crate::thread_executor::install_thread_executor(executor);
    block_on(future)
}

/// Watches for a blocked thread that is never going to be woken.
///
/// The deadlock this exists to catch — parking the thread that was supposed to be
/// running the executor — has a distinctive signature: not "slow", but *zero* wakeups,
/// ever. So the watchdog measures time since the last wake rather than total elapsed
/// time, which keeps it quiet for a future that is merely waiting a long time on
/// something real.
struct StallWatchdog {
    /// `None` once the check is disabled, either by configuration or because it has
    /// already fired and there is no value in repeating it.
    threshold: Option<Duration>,
    last_wake: Instant,
}

impl StallWatchdog {
    fn new() -> Self {
        let secs = std::env::var("SOME_EXECUTOR_BLOCK_ON_STALL_SECS")
            .ok()
            .and_then(|v| v.trim().parse::<u64>().ok())
            .unwrap_or(DEFAULT_STALL_SECS);
        StallWatchdog {
            threshold: (secs > 0).then(|| Duration::from_secs(secs)),
            last_wake: Instant::now(),
        }
    }

    /// How long the thread may wait before the watchdog wants to look again, or `None`
    /// to wait indefinitely because the check is off.
    fn budget(&self) -> Option<Duration> {
        self.threshold
            .map(|t| t.saturating_sub(self.last_wake.elapsed()))
    }

    /// Records that the future was woken, so the thread is demonstrably not deadlocked.
    fn woken(&mut self) {
        if self.threshold.is_some() {
            self.last_wake = Instant::now();
        }
    }

    /// Whether the thread has now waited past the threshold without a wakeup.
    ///
    /// Returns `false` for a spurious wakeup that arrived early, and `false` forever
    /// after the first `true`: the diagnostic is worth saying once, not on a loop.
    fn stalled(&mut self) -> bool {
        if self.budget() != Some(Duration::ZERO) {
            return false;
        }
        self.threshold = None;
        true
    }
}

/// Off wasm32: park on a condvar between polls.
#[cfg(not(target_arch = "wasm32"))]
mod imp {
    use super::StallWatchdog;
    use std::future::Future;
    use std::pin::pin;
    use std::sync::{Arc, Condvar, Mutex};
    use std::task::{Context, Poll, Wake, Waker};

    /// A wake flag plus the condvar the polling thread waits on.
    ///
    /// The flag is *sticky*: `wake` sets it whether or not anyone is waiting yet, so a
    /// wake delivered from inside `poll` — before the thread reaches `wait` — is still
    /// there when it gets there. Without that, a future woken by its own poll would
    /// hang.
    struct Signal {
        woken: Mutex<bool>,
        condvar: Condvar,
    }

    /// Poisoning means a panic happened while the flag was held — from the polled
    /// future, or from the stall diagnostic escalating to a panic. Neither can leave a
    /// `bool` in a state worth refusing to read, and honoring the poison would turn one
    /// panic into a second, confusing one on whichever thread next calls `wake`.
    fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
        mutex.lock().unwrap_or_else(|e| e.into_inner())
    }

    impl Wake for Signal {
        fn wake(self: Arc<Self>) {
            self.wake_by_ref();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            let mut woken = lock(&self.woken);
            *woken = true;
            self.condvar.notify_one();
        }
    }

    pub(super) fn block_on<F: Future>(future: F) -> F::Output {
        let signal = Arc::new(Signal {
            woken: Mutex::new(false),
            condvar: Condvar::new(),
        });
        let waker = Waker::from(signal.clone());
        let mut context = Context::from_waker(&waker);
        let mut future = pin!(future);
        let mut watchdog = StallWatchdog::new();

        loop {
            if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
                return output;
            }

            let mut woken = lock(&signal.woken);
            while !*woken {
                match watchdog.budget() {
                    Some(budget) => {
                        let (guard, timeout) = signal
                            .condvar
                            .wait_timeout(woken, budget)
                            .unwrap_or_else(|e| e.into_inner());
                        woken = guard;
                        if timeout.timed_out() && !*woken && watchdog.stalled() {
                            crate::warn_or_panic(super::STALL_MESSAGE);
                        }
                    }
                    None => {
                        woken = signal
                            .condvar
                            .wait(woken)
                            .unwrap_or_else(|e| e.into_inner());
                    }
                }
            }
            *woken = false;
            drop(woken);
            watchdog.woken();
        }
    }
}

/// On wasm32: poll, then yield the thread, and repeat.
///
/// A condvar is the wrong shape here even on a worker. A wasm future is as likely to be
/// waiting on that worker's *event loop* — a timer, a settled promise — as on another
/// thread, and the event loop cannot run while the worker sits in `atomic.wait` on a
/// condvar nobody will notify. Yielding with a short timeout instead keeps the loop
/// making progress either way. The waker is still installed, because the watchdog needs
/// to distinguish "nothing has woken this" from "slow".
#[cfg(target_arch = "wasm32")]
mod imp {
    use super::{MAIN_THREAD_MESSAGE, StallWatchdog};
    use std::future::Future;
    use std::pin::pin;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::task::{Context, Poll, Wake, Waker};

    /// Sticky wake flag, for the same reason the native arm's is sticky.
    struct Signal(AtomicBool);

    impl Wake for Signal {
        fn wake(self: Arc<Self>) {
            self.wake_by_ref();
        }

        fn wake_by_ref(self: &Arc<Self>) {
            self.0.store(true, Ordering::Release);
        }
    }

    pub(super) fn block_on<F: Future>(future: F) -> F::Output {
        if wasm_lite_std::is_main_thread() {
            panic!("{}", MAIN_THREAD_MESSAGE);
        }

        let signal = Arc::new(Signal(AtomicBool::new(false)));
        let waker = Waker::from(signal.clone());
        let mut context = Context::from_waker(&waker);
        let mut future = pin!(future);
        let mut watchdog = StallWatchdog::new();

        loop {
            if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
                return output;
            }
            if signal.0.swap(false, Ordering::Acquire) {
                watchdog.woken();
            } else if watchdog.stalled() {
                crate::warn_or_panic(super::STALL_MESSAGE);
            }
            // Yield even when the future just woke itself. Re-polling immediately
            // would be lower latency, but a future that wakes itself every poll
            // would then never let this worker's event loop turn -- and the event
            // loop is where most wasm wakeups come from, so that trades a 1ms delay
            // for a livelock. Traps on the main thread, which is rejected above.
            wasm_lite_std::yield_now();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::block_on;
    use crate::sys::Instant;
    use std::future::Future;
    use std::pin::Pin;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::task::{Context, Poll};

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn ready_future_returns_immediately() {
        assert_eq!(block_on(async { 42 }), 42);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn borrows_from_the_stack() {
        // The point of polling in place rather than spawning: no Send, no 'static.
        let name = String::from("world");
        assert_eq!(
            block_on(async { format!("hello {}", name.as_str()) }),
            "hello world"
        );
        assert_eq!(
            name, "world",
            "the future borrowed, it did not take ownership"
        );
    }

    /// Returns `Pending` a fixed number of times, waking itself each time.
    ///
    /// Self-waking from inside `poll` is the case the sticky wake flag exists for: the
    /// wake lands before the thread reaches its wait.
    struct SelfWaking(usize);

    impl Future for SelfWaking {
        type Output = usize;

        fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<usize> {
            if self.0 == 0 {
                return Poll::Ready(0);
            }
            self.0 -= 1;
            cx.waker().wake_by_ref();
            Poll::Pending
        }
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn self_waking_future_completes() {
        assert_eq!(block_on(SelfWaking(5)), 0);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn polls_only_after_wake() {
        // A future that never wakes must not be re-polled in a spin; count polls
        // across a self-wake sequence to confirm the loop is wake-driven.
        struct Counting(Arc<AtomicUsize>, usize);
        impl Future for Counting {
            type Output = ();
            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
                self.0.fetch_add(1, Ordering::Relaxed);
                if self.1 == 0 {
                    return Poll::Ready(());
                }
                self.1 -= 1;
                cx.waker().wake_by_ref();
                Poll::Pending
            }
        }
        let polls = Arc::new(AtomicUsize::new(0));
        block_on(Counting(polls.clone(), 3));
        assert_eq!(polls.load(Ordering::Relaxed), 4);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn watchdog_fires_once_and_only_after_the_threshold() {
        use super::StallWatchdog;
        use std::time::Duration;

        let mut watchdog = StallWatchdog {
            threshold: Some(Duration::from_millis(50)),
            last_wake: Instant::now(),
        };
        // A wakeup that beats the threshold is not a stall.
        assert!(!watchdog.stalled());

        // Burn past the threshold without recording a wake.
        watchdog.last_wake = Instant::now() - Duration::from_secs(1);
        assert!(watchdog.stalled());
        // ...and the diagnostic is not repeated on every subsequent park.
        assert!(!watchdog.stalled());
        assert_eq!(watchdog.budget(), None);
    }

    #[cfg_attr(not(target_arch = "wasm32"), test)]
    #[cfg_attr(target_arch = "wasm32", wasm_lite::wasm_lite_test(worker))]
    fn watchdog_disabled_waits_indefinitely() {
        use super::StallWatchdog;

        let mut watchdog = StallWatchdog {
            threshold: None,
            last_wake: Instant::now(),
        };
        assert_eq!(watchdog.budget(), None);
        assert!(!watchdog.stalled());
    }

    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn woken_from_another_thread() {
        use std::sync::Mutex;
        use std::task::Waker;
        use std::time::Duration;

        struct Remote(Arc<Mutex<(bool, Option<Waker>)>>);
        impl Future for Remote {
            type Output = &'static str;
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<&'static str> {
                let mut state = self.0.lock().unwrap();
                if state.0 {
                    Poll::Ready("done")
                } else {
                    state.1 = Some(cx.waker().clone());
                    Poll::Pending
                }
            }
        }

        let state = Arc::new(Mutex::new((false, None::<Waker>)));
        let remote = state.clone();
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(50));
            let waker = {
                let mut state = remote.lock().unwrap();
                state.0 = true;
                state.1.take()
            };
            if let Some(waker) = waker {
                waker.wake();
            }
        });

        assert_eq!(block_on(Remote(state)), "done");
    }
}