yash_executor/
lib.rs

1// This file is part of yash, an extended POSIX shell.
2// Copyright (C) 2024 WATANABE Yuki
3
4//! `yash-executor` is a library for running concurrent tasks in a
5//! single-threaded context. This crate supports `no_std` configurations but
6//! requires the `alloc` crate.
7//!
8//! The [`Executor`] provided by this crate can be instantiated more than once
9//! to run multiple sets of tasks concurrently. Each executor maintains its
10//! own set of tasks and does not share tasks with other executors. This is
11//! different from other executor implementations that use a global or
12//! thread-local executor.
13//!
14//! This crate is free of locks and atomic operations at the cost of
15//! [unsafe spawning](Executor::spawn_pinned). Wakers used in this crate are
16//! thread-unsafe and not guarded by locks or atomics, so you must ensure that
17//! wakers are not shared between threads.
18//!
19//! ```
20//! # use yash_executor::Executor;
21//! # use yash_executor::forwarder::TryReceiveError;
22//! let executor = Executor::new();
23//!
24//! // Spawn a task that returns 42
25//! let receiver = unsafe { executor.spawn(async { 42 }) };
26//!
27//! // The task is not yet complete
28//! assert_eq!(receiver.try_receive(), Err(TryReceiveError::NotSent));
29//!
30//! // Run the executor until the task is complete
31//! executor.run_until_stalled();
32//!
33//! // Now we have the result
34//! assert_eq!(receiver.try_receive(), Ok(42));
35//! ```
36//!
37//! [`Spawner`]s provide a subset of the functionality of [`Executor`] to allow
38//! spawning tasks without access to the full executor. It is useful for adding
39//! tasks from within another task without creating cyclic dependencies, which
40//! can cause memory leaks.
41//!
42//! The [`forwarder`] module provides utilities for forwarding the result of a
43//! future to another future. The [`forwarder`](forwarder::forwarder) function
44//! creates a pair of [`Sender`] and [`Receiver`] that share an internal state
45//! to communicate the result of a future. A `Receiver` is also returned from
46//! the [`Executor::spawn`] method to receive the result of a future.
47//!
48//! [`Sender`]: forwarder::Sender
49//! [`Receiver`]: forwarder::Receiver
50
51#![no_std]
52extern crate alloc;
53
54use alloc::boxed::Box;
55use alloc::collections::VecDeque;
56use alloc::rc::{Rc, Weak};
57use core::cell::RefCell;
58use core::fmt::Debug;
59use core::pin::Pin;
60
61/// Interface for running concurrent tasks
62///
63/// You call the [`spawn_pinned`](Self::spawn_pinned) or [`spawn`](Self::spawn)
64/// method to add a task to the executor. Just adding a task to the executor
65/// does not run it. You need to call the [`step`](Self::step) or
66/// [`run_until_stalled`](Self::run_until_stalled) method to run the tasks.
67///
68/// `Executor` implements `Clone` but all clones share the same set of tasks.
69/// Separately created `Executor` instances do not share tasks.
70#[derive(Clone, Debug, Default)]
71pub struct Executor<'a> {
72    state: Rc<RefCell<ExecutorState<'a>>>,
73}
74
75/// Interface for spawning tasks
76///
77/// `Spawner` provides a subset of the functionality of `Executor` to allow
78/// spawning tasks without access to the full executor.
79///
80/// `Spawner` instances can be cloned and share the same executor state.
81/// `Spawner`s maintain a weak reference to the executor state, so they do not
82/// prevent the executor from being dropped. If the executor is dropped, the
83/// `Spawner` will not be able to spawn any more tasks.
84///
85/// To obtain a `Spawner` from an `Executor`, use the [`Executor::spawner`]
86/// method. The [`dead`](Self::dead) and `default` functions return a `Spawner`
87/// that can never spawn tasks.
88///
89/// ```
90/// # use yash_executor::Executor;
91/// let executor = Executor::new();
92/// let spawner = executor.spawner();
93/// let final_receiver = unsafe {
94///     executor.spawn(async move {
95///         let receiver_1 = spawner.spawn(async { 1 }).unwrap();
96///         let receiver_2 = spawner.spawn(async { 2 }).unwrap();
97///         receiver_2.await + receiver_1.await
98///     })
99/// };
100/// executor.run_until_stalled();
101/// assert_eq!(final_receiver.try_receive(), Ok(3));
102/// ```
103#[derive(Clone, Debug, Default)]
104pub struct Spawner<'a> {
105    state: Weak<RefCell<ExecutorState<'a>>>,
106}
107
108/// Internal state of the executor
109#[derive(Default)]
110struct ExecutorState<'a> {
111    /// Queue of woken tasks to be executed
112    ///
113    /// Tasks are added to the queue when they are woken up by another task or
114    /// when they are spawned. The executor removes a task from the queue and
115    /// polls it once. If the poll method returns `Poll::Pending`, the task
116    /// needs to be added back to the queue by some waker when it is ready to
117    /// be polled again.
118    wake_queue: VecDeque<Rc<Task<'a>>>,
119    // We don't need to store tasks that are waiting to be woken up because they
120    // are retained by wakers. This also prevents leaking tasks that are never
121    // woken up.
122}
123
124impl Debug for ExecutorState<'_> {
125    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
126        f.debug_struct("ExecutorState")
127            .field(
128                "wake_queue",
129                &format_args!("(len = {})", self.wake_queue.len()),
130            )
131            .finish()
132    }
133}
134
135/// State of a task to be executed
136struct Task<'a> {
137    /// Shared state of the executor for running this task
138    executor: Weak<RefCell<ExecutorState<'a>>>,
139
140    /// The task to be executed
141    ///
142    /// This value becomes `None` when the task is completed to prevent polling
143    /// it again.
144    future: RefCell<Option<Pin<Box<dyn Future<Output = ()> + 'a>>>>,
145}
146
147pub mod forwarder;
148
149mod executor;
150mod spawner;
151mod task;
152mod waker;
153
154pub use spawner::SpawnError;
yash_executor/lib.rs

yash_executor/
lib.rs
