rstango 0.4.0

//! The pool of threads that blocking work runs on, so that it never blocks the executor
//! polling the future waiting for it: the requests of the served devices, whose handlers
//! block for as long as they run, and the lookups of host names, see
//! [`crate::future::resolve`].
//!
//! The pool grows as it needs to, a request finding no idle thread starting one, and each
//! thread ends after it has been idle for a while. It is not bounded to a few threads: a
//! handler waiting on another device of the same server holds its thread for as long as
//! that other device takes, which would deadlock a pool whose every thread is waiting so.

use std::panic::{self, AssertUnwindSafe};
use std::sync::{Condvar, Mutex};
use std::collections::VecDeque;
use std::task::{Context, Poll};
use std::time::Duration;
use std::future::Future;
use std::pin::Pin;
use std::thread;

use futures::executor::block_on;
use futures::future::BoxFuture;
use futures::channel::oneshot;
use tracing::warn;

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


/// How long a thread of the pool waits for a job before it ends.
const IDLE_TIMEOUT: Duration = Duration::from_secs(10);

/// The most threads the pool runs at once, far above what a process should need, which is
/// only there to keep a flood of requests from starting threads until the system fails to.
const MAX_THREADS: usize = 1024;

type Job = Box<dyn FnOnce() + Send>;

static POOL: Pool = Pool {
    state: Mutex::new(State { jobs: VecDeque::new(), threads: 0, idle: 0, started: 0 }),
    cond: Condvar::new(),
};

struct Pool {
    state: Mutex<State>,
    cond: Condvar,
}

struct State {
    jobs: VecDeque<Job>,
    /// The threads running, idle ones included.
    threads: usize,
    /// The threads waiting for a job.
    idle: usize,
    /// The threads ever started, which is what the next one is numbered with, see
    /// [`spawn`]: an index is never reused, so a thread is told apart in the logs from one
    /// that ended before it.
    started: usize,
}

/// Run that function on the pool, and give back a future for what it returns.
pub fn spawn<T: Send + 'static>(func: impl FnOnce() -> Result<T> + Send + 'static) -> Task<T> {

    let (tx, rx) = oneshot::channel();
    let job: Job = Box::new(move || { let _ = tx.send(func()); });

    let mut state = POOL.state.lock().unwrap();
    state.jobs.push_back(job);

    // Only as many threads are started as there are jobs no idle thread will take.
    if state.idle < state.jobs.len() && state.threads < MAX_THREADS {
        state.threads += 1;
        state.started += 1;
        if let Err(err) = thread::Builder::new().name(format!("pool-{}", state.started)).spawn(worker) {
            state.threads -= 1;
            warn!(%err, "Failed to start a thread of the pool, leaving the job to the running ones");
        }
    }

    drop(state);
    POOL.cond.notify_one();

    Task(rx)

}

/// Run a request, on the pool when it may block, and as-is otherwise.
///
/// A request runs on the pool as soon as one of its handlers is synchronous: that handler
/// may block, so it is given a thread, and the asynchronous handlers of the same request
/// are blocked on there, the whole request being a single job. A request whose handlers
/// are all asynchronous is polled by the executor of the backend instead, holding no
/// thread while it waits.
pub fn run<T>(blocking: bool, request: impl Future<Output = Result<T>> + Send + 'static) -> BoxFuture<'static, Result<T>>
where
    T: Send + 'static,
{
    if blocking {
        Box::pin(spawn(move || block_on(request)))
    } else {
        Box::pin(request)
    }
}

/// The future of a function run on the pool, see [`spawn`].
#[derive(Debug)]
pub struct Task<T>(oneshot::Receiver<Result<T>>);

impl<T> Future for Task<T> {

    type Output = Result<T>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut self.0).poll(cx).map(|res| res.unwrap_or_else(|_| Err(Error::new(ErrorKind::Internal)
            .with_description("Pool dropped the job before it answered"))))
    }

}

/// The loop of a thread of the pool.
fn worker() {
    let mut state = POOL.state.lock().unwrap();
    loop {
        if let Some(job) = state.jobs.pop_front() {
            drop(state);
            // A job that panicked answers nothing, its sender being dropped, and the thread
            // goes on taking jobs.
            let _ = panic::catch_unwind(AssertUnwindSafe(job));
            state = POOL.state.lock().unwrap();
            continue;
        }
        state.idle += 1;
        let (next, timeout) = POOL.cond.wait_timeout(state, IDLE_TIMEOUT).unwrap();
        state = next;
        state.idle -= 1;
        if timeout.timed_out() && state.jobs.is_empty() {
            state.threads -= 1;
            return;
        }
    }
}