zng/

task.rs

//! Parallel async tasks and async task runners.
//!
//! Use [`run`], [`respond`] or [`spawn`] to run parallel tasks, use [`wait`], [`io`] and [`fs`] to unblock
//! IO operations and use [`http`] for async HTTP.
//!
//! All functions of this module propagate the [`LocalContext`].
//!
//! This crate also re-exports the [`rayon`] and [`parking_lot`] crates for convenience. You can use the
//! [`with_ctx`] adapter in rayon iterators to propagate the [`LocalContext`]. You can
//! also use [`join`] to propagate thread context for a raw rayon join operation.
//!
//! [`with_ctx`]: crate::task::rayon::ParallelIteratorExt::with_ctx
//!
//! # Examples
//!
//! ```
//! use zng::prelude::*;
//!
//! # fn example() {
//! let enabled = var(false);
//! # let _ =
//! Button! {
//!     on_click = async_hn!(enabled, |_| {
//!         enabled.set(false);
//!
//!         let sum_task = task::run(async {
//!             let numbers = read_numbers().await;
//!             numbers.par_iter().map(|i| i * i).sum()
//!         });
//!         let sum: usize = sum_task.await;
//!         println!("sum of squares: {sum}");
//!
//!         enabled.set(true);
//!     });
//!     widget::enabled = enabled;
//! }
//! # ; }
//!
//! async fn read_numbers() -> Vec<usize> {
//!     let raw = task::wait(|| std::fs::read_to_string("numbers.txt").unwrap()).await;
//!     raw.par_split(',').map(|s| s.trim().parse::<usize>().unwrap()).collect()
//! }
//! ```
//!
//! The example demonstrates three different ***tasks***, the first is a [`UiTask`] in the `async_hn` handler,
//! this task is *async* but not *parallel*, meaning that it will execute in more then one app update, but it will only execute in the
//! `on_click` context and thread. This is good for coordinating UI state, like setting variables, but is not good if you want to do CPU intensive work.
//!
//! To keep the app responsive we move the computation work inside a [`run`] task, this task is *async* and *parallel*,
//! meaning it can `.await` and will execute in parallel threads. It runs in [`rayon`] so you can
//! easily make the task multi-threaded and when it is done it sends the result back to the widget task that is awaiting for it. We
//! resolved the responsiveness problem, but there is one extra problem to solve, how to not block one of the worker threads waiting IO.
//!
//! We want to keep the [`run`] threads either doing work or available for other tasks, but reading a file is just waiting
//! for a potentially slow external operation, so if we call [`std::fs::read_to_string`] directly we can potentially remove one of
//! the worker threads from play, reducing the overall tasks performance. To avoid this we move the IO operation inside a [`wait`]
//! task, this task is not *async* but it is *parallel*, meaning if does not block but it runs a blocking operation. It runs inside
//! a [`blocking`] thread-pool, that is optimized for waiting.
//!
//! # Async IO
//!
//! You can use [`wait`], [`io`] and [`fs`] to do async IO, Zng uses this API for internal async IO, they are just a selection
//! of external async crates re-exported for convenience and compatibility.
//!
//! The [`io`] module just re-exports the [`futures-lite::io`] traits and types, adding only progress tracking. The
//! [`fs`] module is the [`async-fs`] crate. Most of the IO async operations are implemented using extensions traits
//! so we recommend blob importing [`io`] to start implementing async IO.
//!
//! ```
//! use zng::prelude::*;
//!
//! async fn read_numbers() -> Result<Vec<usize>, Box<dyn std::error::Error + Send + Sync>> {
//!     let mut file = task::fs::File::open("numbers.txt").await?;
//!     let mut raw = String::new();
//!     file.read_to_string(&mut raw).await?;
//!     raw.par_split(',').map(|s| s.trim().parse::<usize>().map_err(Into::into)).collect()
//! }
//! ```
//!
//! All the `std::fs` synchronous operations have an async counterpart in [`fs`]. For simpler one shot
//! operation it is recommended to just use `std::fs` inside [`wait`], the async [`fs`] types are not async at
//! the OS level, they only offload operations inside the same thread-pool used by [`wait`].
//!
//! # HTTP Client
//!
//! You can use [`http`] to implement asynchronous HTTP requests. Zng also uses the [`http`] module for
//! implementing operations such as loading an image from a given URL, the module defines an HTTP client API
//!  that is backend agnostic. By default it uses the system `curl` command
//! line utility with a simple cache, this can be replaced using the full API.
//!
//! ```
//! use zng::prelude::*;
//!
//! # fn example() {
//! let enabled = var(false);
//! let msg = var("loading..".to_txt());
//! # let _ =
//! Button! {
//!     on_click = async_hn!(enabled, msg, |_| {
//!         enabled.set(false);
//!
//!         match task::http::get_txt("https://httpbin.org/get").await {
//!             Ok(r) => msg.set(r),
//!             Err(e) => msg.set(formatx!("error: {e}")),
//!         }
//!
//!         enabled.set(true);
//!     });
//! }
//! # ; }
//! ```
//!
//! For other protocols or alternative HTTP clients you can use [external crates](#async-crates-integration).
//!
//! # Async Crates Integration
//!
//! You can use external async crates to create futures and then `.await` then in async code managed by Zng, but there is some
//! consideration needed. Async code needs a runtime to execute and some async functions from external crates expect their own runtime
//! to work properly, as a rule of thumb if the crate starts their own *event reactor* you can just use then without worry.
//!
//! You can use the [`futures`], [`async-std`] and [`smol`] crates without worry, they integrate well and even use the same [`blocking`]
//! thread-pool that is used in [`wait`]. Functions that require an *event reactor* start it automatically, usually at the cost of one extra
//! thread only. Just `.await` futures from these crate.
//!
//! The [`tokio`] crate on the other hand, does not integrate well. It does not start its own runtime automatically, and expects you
//! to call its async functions from inside the tokio runtime. After you create a future from inside the runtime you can `.await` then
//! in any thread, so we recommend manually starting its runtime in a thread and then using the `tokio::runtime::Handle` to start
//! futures in the runtime.
//!
//! External tasks also don't propagate the thread context, if you want access to app services or want to set vars inside external
//! parallel closures you must capture and load the [`LocalContext`] manually.
//!
//! [`LocalContext`]: crate::app::LocalContext
//! [`blocking`]: https://docs.rs/blocking
//! [`futures`]: https://docs.rs/futures
//! [`async-std`]: https://docs.rs/async-std
//! [`smol`]: https://docs.rs/smol
//! [`tokio`]: https://docs.rs/tokio
//! [`futures-lite::io`]: https://docs.rs/futures-lite/*/futures_lite/io/index.html
//! [`async-fs`]: https://docs.rs/async-fs
//! [`rayon`]: https://docs.rs/rayon
//! [`parking_lot`]: https://docs.rs/parking_lot
//!
//! # Full API
//!
//! See [`zng_task`] for the full API.

pub use zng_task::{
    DeadlineError, McWaker, Progress, ScopeCtx, SignalOnce, TaskPanicError, UiTask, all, all_ok, all_some, any, any_ok, any_some, block_on,
    deadline, fs, future_fn, io, join, join_context, poll_respond, poll_spawn, respond, run, run_catch, scope, set_spawn_panic_handler,
    spawn, spawn_wait, wait, wait_catch, wait_respond, with_deadline, yield_now,
};

#[cfg(any(doc, feature = "test_util"))]
pub use zng_task::{doc_test, spin_on};

/// HTTP client.
///
/// This module provides an HTTP client API that is backend agnostic. By default it uses the system `curl` command
/// line utility with a simple cache, this can be replaced using the full API.
///
/// # Examples
///
/// Get some text:
///
/// ```
/// # use zng::task;
/// # async fn demo() -> Result<(), zng::task::http::Error> {
/// let text = task::http::get_txt("https://httpbin.org/base64/SGVsbG8gV29ybGQ=").await?;
/// println!("{text}!");
/// # Ok(()) }
/// ```
///
/// # Full API
///
/// See [`zng_task::http`] for the full API.
#[cfg(feature = "http")]
pub mod http {
    pub use zng_task::http::{
        CacheMode, Error, Method, Request, Response, StatusCode, Uri, delete, get, get_bytes, get_json, get_txt, head, header, method,
        post, put, send, set_request_default, uri,
    };

    /// Remove all cached entries, or just older ones if `prune` is enabled.
    pub async fn cache_clean(prune: bool) {
        if prune {
            zng_task::http::http_cache().prune().await;
        } else {
            zng_task::http::http_cache().purge().await;
        }
    }
}

/// Communication channels.
///
/// Use [`bounded`], [`unbounded`] and [`rendezvous`] to create channels for use across threads in the same process.
/// Use [`ipc_unbounded`] to create channels that work across processes.
///
/// # Examples
///
/// ```no_run
/// use zng::prelude::*;
///
/// let (sender, receiver) = task::channel::bounded(5);
///
/// task::spawn(async move {
///     task::deadline(5.secs()).await;
///     if let Err(e) = sender.send("Data!").await {
///         eprintln!("no receiver connected, did not send message: '{e}'")
///     }
/// });
/// task::spawn(async move {
///     match receiver.recv().await {
///         Ok(msg) => println!("{msg}"),
///         Err(_) => eprintln!("no message in channel and no sender connected"),
///     }
/// });
/// ```
///
/// [`bounded`]: crate::task::channel::bounded
/// [`unbounded`]: crate::task::channel::unbounded
/// [`rendezvous`]: crate::task::channel::rendezvous
/// [`ipc_unbounded`]: crate::task::channel::ipc_unbounded
///
/// # Full API
///
/// See [`zng_task::channel`] for the full API.
pub mod channel {
    pub use zng_task::channel::{ChannelError, Receiver, Sender, bounded, rendezvous, unbounded};
    pub use zng_task::channel::{
        IpcBytes, IpcBytesCast, IpcBytesCastIntoIter, IpcBytesIntoIter, IpcBytesMut, IpcBytesMutCast, IpcBytesWriter,
        IpcBytesWriterBlocking, IpcFileHandle, IpcRead, IpcReadBlocking, IpcReadHandle, IpcReceiver, IpcSender, IpcValue, NamedIpcReceiver,
        NamedIpcSender, WeakIpcBytes, ipc_unbounded,
    };
}

#[cfg(not(wasm))]
pub use zng_task::process;

pub use zng_app::widget::UiTaskWidget;

/// Recommended blocking locks.
///
/// The `Mutex` and `RwLock` types reexported here are recommended, they are
/// more optimized than `std` alternatives because they don't implement lock poisoning,
/// have `const` initialization and integrate with the `zng` feature `"deadlock_detection"`.
///
/// # Full API
///
/// See the [`parking_lot`] crate for the full API.
///
/// [`parking_lot`]: https://docs.rs/parking_lot
pub mod parking_lot {
    use zng_task::parking_lot;

    // no_inline is not imported from zng_task
    #[doc(no_inline)]
    pub use parking_lot::{
        ArcMutexGuard, ArcRwLockReadGuard, ArcRwLockWriteGuard, Condvar, MappedMutexGuard, MappedRwLockReadGuard, MappedRwLockWriteGuard,
        Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockUpgradableReadGuard, RwLockWriteGuard,
    };
}

/// Parallel iterators.
///
/// This module mostly reexports the primary traits from the [`rayon`] crate.
///
/// This module also includes [`ParallelIteratorWithCtx`] that propagates the
/// zng app context to rayon tasks.
///
/// # Full API
///
/// See the [`rayon`] crate for the full API.
///
/// [`rayon`]: https://docs.rs/rayon
/// [`ParallelIteratorWithCtx`]: crate::task::rayon::ParallelIteratorWithCtx
pub mod rayon {
    pub use zng_task::rayon::{ParallelIteratorExt, ParallelIteratorWithCtx};

    use zng_task::rayon;

    #[doc(no_inline)]
    pub use rayon::{iter, slice, str};

    /// Rayon traits imported `as _`.
    pub mod prelude {
        use zng_task::rayon;

        #[doc(no_inline)]
        pub use rayon::{
            iter::FromParallelIterator as _, iter::IndexedParallelIterator as _, iter::IntoParallelIterator as _,
            iter::IntoParallelRefIterator as _, iter::IntoParallelRefMutIterator as _, iter::ParallelBridge as _,
            iter::ParallelDrainFull as _, iter::ParallelDrainRange as _, iter::ParallelExtend as _, iter::ParallelIterator as _,
            slice::ParallelSlice as _, slice::ParallelSliceMut as _, str::ParallelString as _,
        };

        pub use zng_task::rayon::ParallelIteratorExt as _;
    }
}