some_global_executor 0.1.6

// SPDX-License-Identifier: MIT OR Apache-2.0
//! A thread-pool executor for async Rust that runs the same code on OS threads natively
//! and on web workers on wasm32.
//!
//! ![logo](https://github.com/drewcrawford/some_global_executor/raw/main/art/logo.png)
//!
//! This crate is the reference executor for the [`some_executor`] framework: if you write
//! code against the [`SomeExecutor`] trait and want one real executor to run it
//! everywhere, this is it.
//!
//! # Why this crate exists
//!
//! Rust ships no async executor in std, and no trait for executors to implement, so most
//! async code is written directly against one specific runtime. The [`some_executor`]
//! crate fixes this by defining a small interface between the code that *has* futures and
//! the code that *runs* them: libraries spawn onto "some" executor through a trait, and
//! applications decide which executor that actually is.
//!
//! That framework needs a production-quality executor to point at, and this crate is it.
//! `some_global_executor` implements the [`SomeExecutor`] trait for `Send + 'static`
//! futures with a resizable pool of workers, and can install itself as the process-wide
//! global executor or as a per-thread executor — which is exactly what
//! [`some_executor`]'s `current_executor()` discovery hierarchy looks for. Out of the
//! box, [`some_executor`] falls back to a built-in toy executor that warns when used;
//! installing this crate is the intended way to replace it.
//!
//! # What it does
//!
//! - **One API, two platforms.** On native targets, tasks run on a pool of OS threads fed
//!   by a `crossbeam-channel` queue. On wasm32, the same public API runs tasks on web
//!   workers, with a custom async channel for distribution (workers can't block). The
//!   platform switch is conditional compilation; your code doesn't change.
//! - **Task observation.** Spawning returns an observer from the [`some_executor`]
//!   framework: await it for the result, poll it, detach it, or drop it to request
//!   cooperative cancellation.
//! - **Graceful shutdown.** [`Executor::drain`] blocks until every spawned task has
//!   finished; [`Executor::drain_async`] is the awaitable version for async contexts.
//! - **Runtime resizing.** [`Executor::resize`] grows or shrinks the worker pool while
//!   the executor is running.
//! - **Structured logging.** Internal operations are logged through
//!   [`logwise`](https://github.com/drewcrawford/logwise), with optional diagnostic,
//!   forensic, and performance feature flags. An optional `exfiltrate` feature exposes a
//!   registry of live pools for external debugging tools.
//!
//! # Quick start
//!
//! ```
//! use some_global_executor::Executor;
//! use some_executor::SomeExecutor;
//! use some_executor::task::{Task, Configuration};
//! # if cfg!(target_arch = "wasm32") { return; }
//!
//! // A named pool with 4 worker threads (or web workers on wasm32).
//! let mut executor = Executor::new("my-executor".to_string(), 4);
//!
//! let task = Task::without_notifications(
//!     "example-task".to_string(),
//!     Configuration::default(),
//!     async { 42 }
//! );
//!
//! let observer = executor.spawn(task);
//! // Await the observer for the result, or drop it to cancel the task.
//!
//! // Optionally make this the executor that some_executor's
//! // current_executor() / global spawning resolves to:
//! executor.set_as_global_executor();
//!
//! // Block until all spawned tasks finish.
//! executor.drain();
//! ```
//!
//! # Where it fits in the ecosystem
//!
//! The [`some_executor`] family divides the work like this:
//!
//! - [`some_executor`] — the trait layer. Defines [`Task`],
//!   observers, cancellation, priorities, task-locals, and the global/thread-local
//!   executor registry. No real scheduling of its own beyond a fallback.
//! - **`some_global_executor`** (this crate) — the reference executor for `Send` tasks:
//!   a thread pool natively, web workers on wasm32.
//! - [`some_local_executor`](https://github.com/drewcrawford/some_local_executor) — runs
//!   non-`Send` tasks on the current thread, and can receive tasks from other threads.
//! - [`test_executors`](https://github.com/drewcrawford/test_executors) — toy executors
//!   for unit tests.
//!
//! So: libraries depend on [`some_executor`] and stay executor-agnostic; applications
//! depend on this crate (or another implementation) and install it once at startup.
//!
//! # Alternatives
//!
//! If you aren't committed to the [`some_executor`] interface, the field looks like this:
//!
//! | Crate | What it is | Trade-off vs. this crate |
//! |---|---|---|
//! | [tokio](https://crates.io/crates/tokio) | Full runtime: executor plus async I/O, timers, sync primitives, and a huge ecosystem | The default choice if you need its I/O stack — but your code becomes tokio-specific, and there is no multithreaded wasm32 story. This crate is executor-only and trait-first. |
//! | [async-executor](https://crates.io/crates/async-executor) (smol) | Small, composable executor; bring your own reactor | Closest in spirit (executor without a bundled runtime), but no common spawning trait for libraries, no observer/cancellation model, and no web-worker pool on wasm32. |
//! | [futures-executor](https://crates.io/crates/futures-executor) | The minimal `ThreadPool`/`LocalPool` in the futures crate | Fine for gluing a future into sync code; no task metadata, draining, resizing, or wasm parallelism. |
//! | [wasm-bindgen-futures](https://crates.io/crates/wasm-bindgen-futures) | `spawn_local` onto the browser event loop | The standard wasm answer, but single-threaded and wasm-only. This crate gives you actual parallelism via web workers and the same API natively. |
//! | [rayon](https://crates.io/crates/rayon) | Data-parallelism for synchronous code | Not an async executor at all; complementary rather than competing. |
//! | async-std | Formerly a full alternative runtime | Discontinued (deprecated in 2025); its maintainers point to smol. |
//! | [`some_executor`]'s built-in fallback | Zero-config executor included in the trait crate | Always available so libraries work with no setup, but deliberately not production quality — it exists to be replaced by this crate. |
//!
//! Pick this crate when you want executor-agnostic code (or already have it via
//! [`some_executor`]), need the same spawning code to work natively and on wasm32 with
//! real parallelism, and don't need a bundled I/O reactor. Pick tokio when you're
//! building on its networking stack and wasm isn't a target.
//!
//! # More examples
//!
//! ## Observing task progress
//!
//! Tasks can be observed to monitor their execution state:
//!
//! ```
//! use some_global_executor::Executor;
//! use some_executor::SomeExecutor;
//! use some_executor::task::{Task, Configuration};
//! use some_executor::observer::{Observer, Observation};
//! # if cfg!(target_arch = "wasm32") { return; }
//!
//! let mut executor = Executor::new("observer-example".to_string(), 2);
//!
//! let task = Task::without_notifications(
//!     "monitored-task".to_string(),
//!     Configuration::default(),
//!     async { "result" }
//! );
//!
//! let observer = executor.spawn(task);
//!
//! // Poll the observer to check task state
//! loop {
//!     match observer.observe() {
//!         Observation::Ready(value) => {
//!             println!("Task completed with: {}", value);
//!             break;
//!         }
//!         Observation::Pending => {
//!             // Task still running
//!             std::thread::yield_now();
//!         }
//!         _ => break,
//!     }
//! }
//!
//! executor.drain();
//! ```
//!
//! ## Global executor pattern
//!
//! Set an executor as the global default for the application:
//!
//! ```
//! use some_global_executor::Executor;
//! # if cfg!(target_arch = "wasm32") { return; }
//!
//! // Create and configure the global executor
//! let executor = Executor::new("global".to_string(), num_cpus::get());
//! executor.set_as_global_executor();
//!
//! // Now tasks can be spawned using the global executor from anywhere
//! // in the application without passing executor references
//!
//! # executor.drain();
//! ```
//!
//! ## Dynamic thread pool management
//!
//! Adjust executor capacity based on workload:
//!
//! ```
//! use some_global_executor::Executor;
//!
//! let mut executor = Executor::new("dynamic".to_string(), 2);
//!
//! // Scale up for heavy workload
//! executor.resize(8);
//!
//! // Scale down during idle periods
//! executor.resize(2);
//!
//! executor.drain();
//! ```
//!
//! # Performance considerations
//!
//! - Thread pool sizing: Default to `num_cpus::get()` for CPU-bound work
//! - For I/O-bound tasks, consider using more threads than CPU cores
//! - WASM targets have platform-specific limitations on parallelism
//! - Use `drain_async()` in async contexts to avoid blocking
//!
//! # Logging
//!
//! This crate uses the `logwise` framework for structured logging. Internal operations
//! are logged at various levels for debugging and monitoring:
//!
//! ```
//! # use some_global_executor::Executor;
//! // Executor creation and operations are automatically logged
//! let executor = Executor::new("logged-executor".to_string(), 4);
//! // Logs: "Creating executor with name logged-executor and 4 threads"
//! # executor.drain();
//! ```
//!
//! # Requirements
//!
//! Rust 1.95+ (edition 2024). Native targets need only stable Rust; running the wasm32
//! test suite requires nightly.
#![warn(missing_docs)]

#[cfg(feature = "exfiltrate")]
pub mod exfiltrate_provider;
#[cfg(feature = "exfiltrate")]
pub mod registry;

use some_executor::DynExecutor;
use some_executor::SomeExecutor;
use some_executor::observer::{Observer, ObserverNotified};
use some_executor::task::{DynSpawnedTask, Task};
use std::any::Any;
use std::convert::Infallible;
use std::future::Future;
use std::hash::Hash;
use std::pin::Pin;

#[cfg(feature = "exfiltrate")]
use std::sync::atomic::AtomicUsize;

/// re-exports the `SomeExecutor` crate for convenience.
pub use some_executor;

#[cfg(feature = "exfiltrate")]
#[derive(Debug)]
struct RegistryRegistration {
    id: u64,
    threads: std::sync::Arc<AtomicUsize>,
    alive: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

#[cfg(feature = "exfiltrate")]
impl PartialEq for RegistryRegistration {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id
    }
}

#[cfg(feature = "exfiltrate")]
impl Eq for RegistryRegistration {}

#[cfg(feature = "exfiltrate")]
impl Hash for RegistryRegistration {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.id.hash(state);
    }
}

#[cfg(feature = "exfiltrate")]
impl Drop for RegistryRegistration {
    fn drop(&mut self) {
        self.alive
            .store(false, std::sync::atomic::Ordering::Release);
    }
}

/// A thread pool-based executor for running asynchronous tasks.
///
/// The `Executor` manages a pool of worker threads that execute submitted tasks.
/// It provides both synchronous and asynchronous draining capabilities, allowing
/// you to wait for all tasks to complete before shutdown.
///
/// # Platform Support
///
/// This executor automatically adapts to the target platform:
/// - On standard platforms, it uses OS threads via `crossbeam-channel`
/// - On WASM targets, it uses web workers for parallelism
///
/// # Examples
///
/// ```
/// use some_global_executor::Executor;
///
/// // Create an executor named "worker-pool" with 4 threads
/// let executor = Executor::new("worker-pool".to_string(), 4);
///
/// // Get the executor's name
/// assert_eq!(executor.name(), "worker-pool");
///
/// // Clean up when done
/// executor.drain();
/// ```
#[derive(Debug, PartialEq, Eq, Hash, Clone)]
pub struct Executor {
    imp: sys::Executor,
    name: String,
    /// This pool's row in the debug registry. See [`registry`].
    #[cfg(feature = "exfiltrate")]
    registry_registration: std::sync::Arc<RegistryRegistration>,
}

impl Executor {
    /// Creates a new executor with the specified name and thread count.
    ///
    /// # Arguments
    ///
    /// * `name` - A descriptive name for the executor, used for logging and debugging
    /// * `threads` - The number of worker threads to create in the thread pool
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// // Create an executor with 8 worker threads
    /// let executor = Executor::new("high-performance".to_string(), 8);
    ///
    /// // Clean up
    /// executor.drain();
    /// ```
    ///
    /// # Logging
    ///
    /// This method logs the creation of the executor using the logwise framework,
    /// including the name and thread count for debugging purposes.
    ///
    /// # Panics
    ///
    /// Panics if `threads` is zero. An executor with no workers cannot make
    /// progress and would otherwise cause [`Executor::drain`] to wait forever.
    pub fn new(name: String, threads: usize) -> Self {
        assert!(
            threads > 0,
            "an executor requires at least one worker thread"
        );
        logwise::event!(
            "some_global_executor.created",
            executor_name = local(name.as_str()),
            thread_count = support(threads as u64),
        );
        let name_clone = name.clone();
        let inner = sys::Executor::new(name, threads);
        #[cfg(feature = "exfiltrate")]
        let registry_registration = {
            let live_threads = std::sync::Arc::new(AtomicUsize::new(threads));
            let live_alive = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
            std::sync::Arc::new(RegistryRegistration {
                id: registry::record_created(
                    &name_clone,
                    threads,
                    live_threads.clone(),
                    live_alive.clone(),
                    inner.drain_notify().clone(),
                ),
                threads: live_threads,
                alive: live_alive,
            })
        };
        Executor {
            imp: inner,
            name: name_clone,
            #[cfg(feature = "exfiltrate")]
            registry_registration,
        }
    }

    /// Returns the name of this executor.
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// let executor = Executor::new("my-executor".to_string(), 2);
    /// assert_eq!(executor.name(), "my-executor");
    /// executor.drain();
    /// ```
    pub fn name(&self) -> &str {
        &self.name
    }
    /// Resizes the thread pool to the specified number of threads.
    ///
    /// This method adjusts the number of worker threads in the executor's thread pool.
    /// The exact behavior depends on the underlying platform implementation.
    ///
    /// # Arguments
    ///
    /// * `threads` - The new number of worker threads
    ///
    /// # Panics
    ///
    /// Panics if `threads` is zero.
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// let mut executor = Executor::new("dynamic".to_string(), 2);
    ///
    /// // Increase thread count for heavy workload
    /// executor.resize(8);
    ///
    /// // Later, reduce thread count
    /// executor.resize(4);
    ///
    /// executor.drain();
    /// ```
    pub fn resize(&mut self, threads: usize) {
        assert!(
            threads > 0,
            "an executor requires at least one worker thread"
        );
        self.imp.resize(threads);
        #[cfg(feature = "exfiltrate")]
        self.registry_registration
            .threads
            .store(threads, std::sync::atomic::Ordering::Relaxed);
    }
}

/// A future that completes when all tasks in an executor have finished.
///
/// `ExecutorDrain` is returned by [`Executor::drain_async()`] and implements
/// `Future<Output = ()>`. It polls the executor's internal state to determine
/// when all tasks have completed execution.
///
/// # Examples
///
/// ```
/// # // This example shows async draining but requires an async runtime
/// # // which is not available in doctests, so we use synchronous drain
/// use some_global_executor::Executor;
/// use some_executor::SomeExecutor;
/// use some_executor::task::{Task, Configuration};
/// # if cfg!(target_arch = "wasm32") { return; }
///
/// let mut executor = Executor::new("async-drain".to_string(), 2);
///
/// // Spawn some work
/// let task = Task::without_notifications(
///     "work".to_string(),
///     Configuration::default(),
///     async {
///         // Simulate some work
///         42
///     }
/// );
/// executor.spawn(task);
///
/// // In async context, you would use:
/// // executor.drain_async().await;
/// // Here we use synchronous drain for the example
/// executor.drain();
/// ```
mod drain;
pub(crate) use drain::DrainNotify;
pub use drain::ExecutorDrain;

/// Platform-specific executor implementations
mod sys;
/// Waker implementation for task notification
#[cfg(not(target_arch = "wasm32"))]
mod waker;

/// Internal representation of a spawned task.
///
/// Wraps the platform-specific task implementation and provides
/// a uniform interface for task management across different platforms.
#[derive(Debug)]
struct SpawnedTask {
    imp: sys::SpawnedTask,
}

impl SpawnedTask {
    fn new(
        task: Box<dyn DynSpawnedTask<Infallible>>,
        drain_notify: std::sync::Arc<DrainNotify>,
    ) -> Self {
        Self {
            imp: sys::SpawnedTask::new(task, drain_notify),
        }
    }
}

impl Executor {
    /// Creates a new executor with default settings.
    ///
    /// This creates an executor named "default" with a platform-appropriate
    /// number of threads (determined by the underlying system implementation).
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// let executor = Executor::new_default();
    /// assert_eq!(executor.name(), "default");
    /// executor.drain();
    /// ```
    pub fn new_default() -> Self {
        Self::new("default".to_string(), sys::default_threadpool_size())
    }

    /// Synchronously waits for all tasks in the executor to complete.
    ///
    /// This method blocks the current thread until all spawned tasks have
    /// finished execution. Native callers sleep on a condition variable;
    /// WASM callers yield while checking for completion.
    ///
    /// # Performance
    ///
    /// This method logs performance warnings if the drain operation takes
    /// an unexpectedly long time, which may indicate stuck or long-running tasks.
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    /// use some_executor::SomeExecutor;
    /// use some_executor::task::{Task, Configuration};
    /// # if cfg!(target_arch = "wasm32") { return; }
    ///
    /// let mut executor = Executor::new("sync-drain".to_string(), 2);
    ///
    /// // Spawn a task
    /// let task = Task::without_notifications(
    ///     "quick-task".to_string(),
    ///     Configuration::default(),
    ///     async { 1 + 1 }
    /// );
    /// executor.spawn(task);
    ///
    /// // Block until all tasks complete
    /// executor.drain();
    /// ```
    pub fn drain(self) {
        #[cfg(feature = "logwise-performance")]
        let _interval = logwise::profile!("some_global_executor.drain");
        #[cfg(not(target_arch = "wasm32"))]
        self.imp.drain_notify().wait_for_completion();
        #[cfg(target_arch = "wasm32")]
        loop {
            //Acquire pairs with the Release decrement on task completion, so task side
            //effects are visible once we observe the count at zero
            let running_tasks = self
                .imp
                .drain_notify()
                .running_tasks
                .load(std::sync::atomic::Ordering::Acquire);
            if running_tasks == 0 {
                break;
            }
            std::thread::yield_now();
        }
    }

    /// Returns a future that completes when all tasks have finished.
    ///
    /// Unlike [`drain()`](Self::drain), this method returns immediately with
    /// an [`ExecutorDrain`] future that can be awaited. This allows for
    /// non-blocking waiting in async contexts.
    ///
    /// # Examples
    ///
    /// ```
    /// # // This example demonstrates async draining
    /// use some_global_executor::Executor;
    /// use some_executor::SomeExecutor;
    /// use some_executor::task::{Task, Configuration};
    /// # if cfg!(target_arch = "wasm32") { return; }
    ///
    /// let mut executor = Executor::new("async-example".to_string(), 4);
    ///
    /// // Spawn multiple tasks
    /// for i in 0..5 {
    ///     let task = Task::without_notifications(
    ///         format!("task-{}", i),
    ///         Configuration::default(),
    ///         async move { i * 2 }
    ///     );
    ///     executor.spawn(task);
    /// }
    ///
    /// // Get the drain future (would be awaited in async context)
    /// let drain_future = executor.drain_async();
    /// // In async context: drain_future.await;
    /// // For this example, convert back to executor and drain synchronously
    /// let executor: Executor = drain_future.into();
    /// executor.drain();
    /// ```
    pub fn drain_async(self) -> ExecutorDrain {
        ExecutorDrain {
            executor: self,
            waiter_id: None,
        }
    }

    fn spawn_internal(&mut self, task: Box<dyn DynSpawnedTask<Infallible>>) {
        let drain_notify = self.imp.drain_notify().clone();
        self.imp
            .spawn_internal(SpawnedTask::new(task, drain_notify));
    }

    /// Sets this executor as the global executor.
    ///
    /// After calling this method, this executor will be used as the default
    /// for global task spawning operations throughout the application.
    /// This is useful for libraries and applications that need a shared
    /// executor instance.
    ///
    /// # Thread Safety
    ///
    /// The executor is cloned when set as global, so the original instance
    /// remains independent and can still be used directly.
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// let executor = Executor::new("global".to_string(), 4);
    /// executor.set_as_global_executor();
    ///
    /// // The global executor is now available for use by any code
    /// // that calls some_executor::global_executor::spawn()
    ///
    /// # executor.drain();
    /// ```
    pub fn set_as_global_executor(&self) {
        some_executor::global_executor::set_global_executor(self.clone_box());
    }

    /// Sets this executor as the thread-local executor.
    ///
    /// After calling this method, this executor will be used as the default
    /// for thread-local task spawning operations on the current thread.
    /// This is useful when different threads need different executor
    /// configurations.
    ///
    /// # Thread Safety
    ///
    /// The thread-local executor is only accessible from the thread that
    /// set it. Each thread can have its own thread-local executor.
    ///
    /// # Examples
    ///
    /// ```
    /// use some_global_executor::Executor;
    ///
    /// let executor = Executor::new("thread-local".to_string(), 2);
    /// executor.set_as_thread_executor();
    ///
    /// // This thread now uses this executor for thread-local spawning
    /// // via some_executor::thread_executor::spawn()
    ///
    /// # executor.drain();
    /// ```
    pub fn set_as_thread_executor(&self) {
        some_executor::thread_executor::set_thread_executor(self.clone_box());
    }
}

/// Implementation of the [`SomeExecutor`] trait from the `some_executor` framework.
///
/// This implementation provides the core task spawning functionality, supporting
/// both synchronous and asynchronous task submission with type-safe observers
/// for monitoring task execution.
impl some_executor::SomeExecutor for Executor {
    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,
    {
        let (spawned, observer) = task.spawn(self);
        self.spawn_internal(Box::new(spawned));
        observer
    }

    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,
    {
        let (spawned, observer) = task.spawn(self);
        self.spawn_internal(Box::new(spawned));
        observer
    }

    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 = some_executor::observer::FinishedObservation<Box<dyn Any + Send>>,
            > + Send,
    > {
        let (spawned, observer) = task.spawn_objsafe(self);
        self.spawn_internal(Box::new(spawned));
        Box::new(observer)
    }

    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 = some_executor::observer::FinishedObservation<
                                Box<dyn Any + Send>,
                            >,
                        > + Send,
                >,
            > + 's,
    > {
        #[allow(clippy::async_yields_async)]
        Box::new(async move { self.spawn_objsafe(task) })
    }

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

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

// Conversion implementations for ergonomic API usage

#[cfg(test)]
mod tests;