Struct Executor

pub struct Executor<'a> { /* private fields */ }

A single-threaded, single-future async executor.

Typical setup

• First, an event mask is created with Events::default() (or Events::new() if that needs to be const). Event masks are Send + Sync and shared references to them are enough for all operations, so they work well as static items or other types of shared state. Distribute event mask references to the external event sources and keep one for the executor.

• The event mask is then watched with Events::watch(). The resulting Signals is Send + !Sync. This means that operations become limited to one thread of execution from this point on, so this is usually done in some type of initialization or main function instead of in a shared or global context.

• A new executor is bound with Signals::bind(). Executors are !Send + !Sync: neither it nor the associated Signals may escape the current thread. This makes them appropriate for construction at the use site.

• A future is created. It needs a reference to the Signals object in order to drive poll functions, making it !Sync too.

• Finally, Executor::block_on() blocks and resolves the future while external event sources direct it through the event mask, possibly with help from the park function.

Examples

This is a complete usage example. It uses std::sync primitives and a park function based on std::thread::park() to multiply the integers from 1 to 10 read from a blocking queue.

use std::{thread, sync::{mpsc::*, Arc}};

bitflags! {
    struct Ev: u32 {
        const QUEUE = 1 << 0;
    }
}

impl EventMask for Ev {
    fn as_bits(self) -> u32 {
        self.bits()
    }
}

async fn recv(signals: &Signals<'_, Ev>, rx: &Receiver<u32>) -> Option<u32> {
    signals.drive_infallible(Ev::QUEUE, || match rx.try_recv() {
        Ok(n) => Ok(Some(n)),
        Err(TryRecvError::Disconnected) => Ok(None),
        Err(TryRecvError::Empty) => Err(nb::Error::WouldBlock),
    }).await
}

let events = Arc::new(Events::default());
let signals = events.watch();

let (tx, rx) = sync_channel(1);
let future = async {
    let mut product = 1;
    while let Some(n) = recv(&signals, &rx).await {
        product *= n;
    }

    product
};

let events_prod = Arc::clone(&events);
let runner = thread::current();

thread::spawn(move || {
    for n in 1..=10 {
        tx.send(n).unwrap();
        events_prod.raise(Ev::QUEUE);
        runner.unpark();
    }

    // Notify shutdown
    drop(tx);
    events_prod.raise(Ev::QUEUE);
    runner.unpark();
});

let result = signals.bind().block_on(future, |park| {
    // thread::park() is event-safe, no lock is required
    let parked = park.race_free();
    if parked.is_idle() {
        thread::park();
    }

    parked
});

assert_eq!(result, (1..=10).product()); // 3628800

Implementations

impl<'exec> Executor<'exec>

pub fn with_waker(self, waker: Waker) -> Self

Replace the executor’s waker with a custom one.

No restrictions are imposed on the waker: nb-executor does not use wakers at all. Application code may define some communication between it and the park function.

pub fn step<'fut, F, P>( self, future: Pin<&'fut mut F>, park: P, ) -> Step<'_, '_, F, P>

where F: Future, P: FnMut(Park<'_>) -> Parked<'_>, 'exec: 'fut,

Begin stepped execution.

This allows the caller to remain in control of program flow in between polls, unlike Executor::block_on(). The future is run as specified in Step and Signals. park is a park function and must follow the Park protocol. The caller and the park function may cooperate to block or sleep outside of the park function within the protocol’s requirements.

pub fn block_on<F, P>(self, future: F, park: P) -> F::Output

where F: Future, P: FnMut(Park<'_>) -> Parked<'_>,

Execute a future on this executor, parking when no progress is possible.

This method will block until the future resolves. There are two possible states of operation while the future is executed:

• Polling: The future’s poll() method is called in order to attempt to resolve it. The signal state is prepared as documented in Signals when switching to the polling state. The next state after polling is unspecified, but will eventually lead to parking if the future pends consistently.

• Parking: This state is entered when useful work is unlikely at the current time. For details, see the parking protocol in Park. park must adhere to this protocol.

See also Step and Executor::step().

pub fn block_busy_on<F: Future>(self, future: F) -> F::Output

Execute a future in a busy-waiting loop.

This is equivalent to calling Executor::block_on() with a park function that never sleeps. This is most likely the wrong way to do whatever you intend, prefer to define a proper wake function.