rstango 0.4.0

//! An extension of [`futures`] for what this crate needs and it lacks.

use std::task::{Context, Poll, Waker};
use std::net::{IpAddr, ToSocketAddrs};
use std::time::{Duration, Instant};
use std::thread::{self, Thread};
use std::sync::{Arc, Mutex};
use std::future::Future;
use std::pin::pin;

use crate::error::{Error, ErrorKind, Result, reason};
use crate::pool;


/// Drive a future to its answer on the calling thread, giving up once the timeout has run
/// out, which is how the blocking methods of this crate are built on the asynchronous ones.
///
/// Running out is the future being **dropped**, and dropping a request is what cancels it,
/// see [`crate::backend::BackendDev`]: nothing is left running behind a call that timed
/// out, and no timer of any executor is needed for this, the calling thread being the one
/// that is parked.
///
/// # Where this comes from
///
/// This is the technique [`futures::executor::block_on`] is built on, and not a copy of it:
/// the waker handed to the future unparks the thread that is blocked on it, so a wake from
/// whichever thread answers the request is what makes this one poll again. What is added
/// here is the deadline, through [`thread::park_timeout`] rather than [`thread::park`],
/// which is the whole reason this is written out rather than `block_on` being called:
/// `block_on` waits forever, and a caller that gives up has to *drop* the future to cancel
/// the request behind it, which a call that never comes back cannot do.
///
/// The real `block_on` keeps its waker in a thread-local and reuses it across calls, which
/// this does not bother with: a request is a round trip over a network, so allocating one
/// [`Arc`] for it is not what this costs.
pub fn block_on_timeout<F: Future>(fut: F, timeout: Duration) -> Result<F::Output> {

    /// Waking this future is unparking the thread that is blocked on it, which is the
    /// whole of the executor a blocking call needs.
    struct ThreadWaker(Thread);

    impl futures::task::ArcWake for ThreadWaker {
        fn wake_by_ref(arc_self: &Arc<Self>) {
            arc_self.0.unpark();
        }
    }

    let mut fut = pin!(fut);
    let waker = futures::task::waker(Arc::new(ThreadWaker(thread::current())));
    let mut cx = Context::from_waker(&waker);
    let deadline = Instant::now() + timeout;

    loop {

        if let Poll::Ready(output) = fut.as_mut().poll(&mut cx) {
            return Ok(output);
        }

        // A park can come back without anything having woken it, so what is left of the
        // deadline is what says whether this is over, and not the park coming back.
        let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
            return Err(Error::new(ErrorKind::TimedOut)
                .with_description(format!("Operation timed out after {timeout:?}")));
        };

        thread::park_timeout(remaining);

    }

}

/// Resolve a host into the socket addresses to try, in the order to try them in.
///
/// A host that is already an address is not looked up at all, which is most of what an IOR
/// carries: only a real name costs anything here. A caller that gives up drops this, which
/// the lookup finds out about when it answers into it.
///
/// # Why a thread
///
/// There is no such thing as an asynchronous `getaddrinfo`. It is the one call that honours
/// `/etc/hosts`, nsswitch, the search domains and whatever else the system was told to
/// resolve with, and it blocks. Every runtime that offers an asynchronous lookup runs that
/// very call on a thread of its own — `tokio::net::lookup_host` is `spawn_blocking` around
/// the standard library's — and so does this one, on the pool of this crate, see
/// [`crate::pool`].
pub async fn resolve(host: &str) -> Result<Vec<IpAddr>> {

    if let Ok(ip) = host.parse::<IpAddr>() {
        return Ok(vec![ip]);
    }

    let owned_host = host.to_string();
    pool::spawn(move || {
        (owned_host.as_str(), 0).to_socket_addrs()
            .map(|addrs| addrs.map(|addr| addr.ip()).collect::<Vec<_>>())
            .map_err(|err| Error::new(ErrorKind::Reason(reason::API_CantResolveHost))
                .with_description(format!("Failed to resolve the IP address of host {owned_host:?}"))
                .with_source(Box::new(err)))
    }).await

}

/// Where the values of one taker are handed over, which a producer may hand to as many
/// takers as it has, one slot each.
///
/// Only the latest value is kept: what goes through a slot is state and not a history, so
/// a taker that does not keep up skips to the latest one. The end of the sequence is kept
/// apart, so that it never replaces the last value, which is taken before it.
pub struct Slot<T> {
    inner: Mutex<SlotInner<T>>,
}

struct SlotInner<T> {
    /// The latest value not taken yet, replacing any before it.
    pending: Option<T>,
    /// Set once the producer is done, see [`Slot::end`].
    ended: bool,
    waker: Option<Waker>,
}

impl<T> Slot<T> {

    pub fn new() -> Self {
        Self {
            inner: Mutex::new(SlotInner {
                pending: None,
                ended: false,
                waker: None,
            }),
        }
    }

    /// Hand a value over, replacing the one that was not taken yet, if any.
    pub fn push(&self, value: T) {

        let waker = {
            let mut inner = self.inner.lock().unwrap();
            inner.pending = Some(value);
            inner.waker.take()
        };

        if let Some(waker) = waker {
            waker.wake();
        }

    }

    /// Tell the taker that no value comes anymore, once it took the one that was pending,
    /// if any.
    pub fn end(&self) {

        let waker = {
            let mut inner = self.inner.lock().unwrap();
            inner.ended = true;
            inner.waker.take()
        };

        if let Some(waker) = waker {
            waker.wake();
        }

    }

    /// Whether the producer is done, see [`Self::end`].
    pub fn is_ended(&self) -> bool {
        self.inner.lock().unwrap().ended
    }

    /// Take the pending value, or register to be woken up when there is one, none coming
    /// once the producer is done and the last value was taken, for as long as it is asked.
    pub fn poll_take(&self, cx: &mut Context<'_>) -> Poll<Option<T>> {
        let mut inner = self.inner.lock().unwrap();
        match inner.pending.take() {
            Some(value) => Poll::Ready(Some(value)),
            None if inner.ended => Poll::Ready(None),
            None => {
                inner.waker = Some(cx.waker().clone());
                Poll::Pending
            }
        }
    }

}