use std::{future::Future, pin::Pin, sync::OnceLock};
    use tokio::{
        runtime::{Handle, RuntimeFlavor},
        select, task,
    };

    /// optionally configured by caller, otherwise falls back to default based on runtime
    pub static EXPERIMENTAL_CPU_HEAVY_TOKIO_EXECUTOR_STRATEGY: OnceLock<CpuHeavyFutureExecutor> =
        OnceLock::new();

    pub type CustomExecutorClosureInput = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;

    pub type CustomExecutorClosure = dyn Fn(CustomExecutorClosureInput) + Send + Sync;

    pub enum CpuHeavyFutureExecutor {
        SpawnBlocking,
        BlockInPlace,
        Custom(Box<CustomExecutorClosure>),
    }

    impl CpuHeavyFutureExecutor {
        /// Send the future to a blocking tokio thread. By default, tokio will spin up a blocking thread
        /// per task, which may be more than your count of CPU cores, depending on runtime config.
        /// If you expect many concurrent cpu-heavy futures, consider limiting your blocking tokio threadpool
        /// size or using `CpuHeavyFutureExecutor::custom()` with a different threadpool.
        pub fn spawn_blocking() -> Self {
            CpuHeavyFutureExecutor::SpawnBlocking
        }

        /// Calls task::block_in_place on the current worker, and evicts other tasks on same worker thread
        /// to avoid blocking them. Can starve your executor of worker threads if called with too many
        /// concurrent cpu-heavy futures.
        pub fn block_in_place() -> Self {
            CpuHeavyFutureExecutor::BlockInPlace
        }

        /// Accepts a closure that will execute the provided future on a background task or different
        /// threadpool, and immediately return. `CpuHeavyFutureExecutor` will use a oneshot
        /// channel to await the result of the provided future, in the foreground.
        pub fn custom(custom_executor_closure: Box<CustomExecutorClosure>) -> Self {
            CpuHeavyFutureExecutor::Custom(custom_executor_closure)
        }
    }

    pub async fn spawn_compute_heavy_future<F, R>(fut: F) -> Result<R, String>
    where
        F: Future<Output = R> + Send + 'static,
        R: Send + 'static,
    {
        let executor = EXPERIMENTAL_CPU_HEAVY_TOKIO_EXECUTOR_STRATEGY
            .get()
            .unwrap_or_else(|| match Handle::current().runtime_flavor() {
                RuntimeFlavor::CurrentThread => &CpuHeavyFutureExecutor::SpawnBlocking,
                _ => &CpuHeavyFutureExecutor::BlockInPlace,
            });
        match executor {
            CpuHeavyFutureExecutor::BlockInPlace => execute_block_in_place(fut).await,
            CpuHeavyFutureExecutor::SpawnBlocking => execute_spawn_blocking(fut).await,
            CpuHeavyFutureExecutor::Custom(custom_executor_closure) => {
                send_to_custom_executor(fut, Box::new(custom_executor_closure)).await
            }
        }
    }

    async fn send_to_custom_executor<T: Send + 'static>(
        future: impl Future<Output = T> + Send + 'static,
        custom_executor_closure: Box<CustomExecutorClosure>,
    ) -> Result<T, String> {
        let (mut tx, rx) = tokio::sync::oneshot::channel();
        custom_executor_closure(Box::pin(async move {
            select!(
                _ = tx.closed() => {
                    // receiver already dropped, don't need to do anything
                }
                result = future => {
                    // if this fails, the receiver already dropped, so we don't need to do anything
                    let _ = tx.send(result);
                }
            )
        }));

        rx.await
            .map_err(|err| format!("error awaiting response from threadpool: {err}"))
    }

    async fn execute_spawn_blocking<F, R>(fut: F) -> Result<R, String>
    where
        F: Future<Output = R> + Send + 'static,
        R: Send + 'static,
    {
        task::spawn_blocking(move || Handle::current().block_on(async { fut.await }))
            .await
            .map_err(|err| format!("error awaiting spawn_blocking handle: {err}"))
    }

    async fn execute_block_in_place<F, R>(fut: F) -> Result<R, String>
    where
        F: Future<Output = R> + Send + 'static,
        R: Send + 'static,
    {
        Ok(task::block_in_place(move || {
            Handle::current().block_on(async { fut.await })
        }))
    }

use std::{future::Future, pin::Pin, sync::OnceLock};

use tokio::{runtime::Handle, sync::mpsc::error::TrySendError};

use compute_heavy_executor_experimental::{
    CpuHeavyFutureExecutor, CustomExecutorClosureInput,
    EXPERIMENTAL_CPU_HEAVY_TOKIO_EXECUTOR_STRATEGY,
};

static CPU_HEAVY_THREAD_POOL: OnceLock<
    tokio::sync::mpsc::Sender<Pin<Box<dyn Future<Output = ()> + Send + 'static>>>,
> = OnceLock::new();

#[tokio::main]
async fn main() {
    // spin up our background threadpool + start a tokio executor on it, listening on our mpsc channel in our oncelock
    init_custom_executor();
    let closure = Box::new(move |fut| custom_executor_closure(fut));
    let compute_heavy_executor = CpuHeavyFutureExecutor::custom(Box::new(closure));
    EXPERIMENTAL_CPU_HEAVY_TOKIO_EXECUTOR_STRATEGY
        .set(compute_heavy_executor)
        .unwrap_or_else(|_| {
            panic!("EXPERIMENTAL_CPU_HEAVY_TOKIO_EXECUTOR_STRATEGY already initialized")
        });
}

fn init_custom_executor() {
    std::thread::Builder::new()
        .name("cpu-heavy-threadpool".to_string())
        .spawn(move || {
            let rt = tokio::runtime::Builder::new_multi_thread()
                .thread_name("cpu-heavy-pool-thread")
                .worker_threads(num_cpus::get() as usize)
                // ref: https://github.com/tokio-rs/tokio/issues/4941
                // consider uncommenting if seeing heavy task contention
                // .disable_lifo_slot()
                .on_thread_start(move || unsafe {
                    // Reduce thread pool thread niceness, so they are lower priority
                    // than the foreground executor and don't interfere with I/O tasks
                    #[cfg(target_os = "linux")]
                    {
                        *libc::__errno_location() = 0;
                        if libc::nice(10) == -1 && *libc::__errno_location() != 0 {
                            let error = std::io::Error::last_os_error();
                            tracing::log::error!("failed to set threadpool niceness: {}", error);
                        }
                    }
                })
                .enable_all()
                .build()
                .unwrap_or_else(|e| panic!("cpu heavy runtime failed_to_initialize: {}", e));
            rt.block_on(async {
                tracing::log::debug!("starting background cpu work");
                process_cpu_work().await;
            });
        })
        .unwrap_or_else(|e| panic!("cpu heavy thread failed_to_initialize: {}", e));
}

async fn process_cpu_work() {
    let (tx, mut rx) = tokio::sync::mpsc::channel(10);
    CPU_HEAVY_THREAD_POOL.set(tx).unwrap();

    while let Some(work) = rx.recv().await {
        tokio::task::spawn(work);
    }
}

fn custom_executor_closure(fut: CustomExecutorClosureInput) {
    let tx = CPU_HEAVY_THREAD_POOL
        .get()
        .expect("Call process_cpu_work() before using cpu heavy threadpool")
        .clone();

    match tx.try_send(Box::pin(fut)) {
        Ok(_) => (),
        Err(TrySendError::Closed(_)) => {
            panic!("background cpu heavy threadpool channel is closed")
        }
        Err(TrySendError::Full(msg)) => {
            tracing::log::warn!("background channel is full, task spawning loop delayed");
            Handle::current().spawn(async move {
                tx.send(msg)
                    .await
                    .expect("background cpu heavy threadpool channel is closed")
            });
        }
    }
}