Struct cros_asyncv2::Executor

source · [−]

pub struct Executor {
    shared: Arc<Mutex<Shared>>,
}

Expand description

An executor for scheduling tasks that poll futures to completion.

All asynchronous operations must run within an executor, which is capable of spawning futures as tasks. This executor also provides a mechanism for performing asynchronous I/O operations.

The returned type is a cheap, clonable handle to the underlying executor. Cloning it will only create a new reference, not a new executor.

Examples

Concurrently wait for multiple files to become readable/writable and then read/write the data.

use std::{
    cmp::min,
    convert::TryFrom,
    fs::OpenOptions,
};

use anyhow::Result;
use cros_async::{Executor, File};
use futures::future::join3;

const CHUNK_SIZE: usize = 32;

// Transfer `len` bytes of data from `from` to `to`.
async fn transfer_data(from: File, to: File, len: usize) -> Result<usize> {
    let mut rem = len;
    let mut buf = [0u8; CHUNK_SIZE];
    while rem > 0 {
        let count = from.read(&mut buf, None).await?;

        if count == 0 {
            // End of file. Return the number of bytes transferred.
            return Ok(len - rem);
        }

        to.write_all(&buf[..count], None).await?;

        rem = rem.saturating_sub(count);
    }

    Ok(len)
}

    let (rx, tx) = base::pipe(true)?;
    let zero = File::open("/dev/zero")?;
    let zero_bytes = CHUNK_SIZE * 7;
    let zero_to_pipe = transfer_data(
        zero,
        File::try_from(tx.try_clone()?)?,
        zero_bytes,
    );

    let rand = File::open("/dev/urandom")?;
    let rand_bytes = CHUNK_SIZE * 19;
    let rand_to_pipe = transfer_data(
        rand,
        File::try_from(tx)?,
        rand_bytes
    );

    let null = OpenOptions::new().write(true).open("/dev/null")?;
    let null_bytes = zero_bytes + rand_bytes;
    let pipe_to_null = transfer_data(
        File::try_from(rx)?,
        File::try_from(null)?,
        null_bytes
    );

    Executor::new().run_until(join3(
        async { assert_eq!(pipe_to_null.await.unwrap(), null_bytes) },
        async { assert_eq!(zero_to_pipe.await.unwrap(), zero_bytes) },
        async { assert_eq!(rand_to_pipe.await.unwrap(), rand_bytes) },
    ))?;

Fields

shared: Arc<Mutex<Shared>>

Implementations

source

Spawn a new future for this executor to run to completion. Callers may use the returned Task to await on the result of f. Dropping the returned Task will cancel f, preventing it from being polled again. To drop a Task without canceling the future associated with it use [Task::detach]. To cancel a task gracefully and wait until it is fully destroyed, use [Task::cancel].

Examples

      let task = ex.spawn(async { 7 + 13 });

      let result = ex.run_until(task)?;
      assert_eq!(result, 20);

source

pub fn spawn_local<F>(&self, f: F) -> Task<F::Output>where
F: Future + 'static,
F::Output: 'static,

Spawn a thread-local task for this executor to drive to completion. Like spawn but without requiring Send on F or F::Output. This method should only be called from the same thread where run() or run_until() is called.

Panics

Executor::run and Executor::run_util will panic if they try to poll a future that was added by calling spawn_local from a different thread.

Examples

      let task = ex.spawn_local(async { 7 + 13 });

      let result = ex.run_until(task)?;
      assert_eq!(result, 20);

source

pub fn spawn_blocking<F, R>(&self, f: F) -> Task<R>where
F: FnOnce() -> R + Send + 'static,
R: Send + 'static,

Run the provided closure on a dedicated thread where blocking is allowed.

Callers may await on the returned Task to wait for the result of f. Dropping or canceling the returned Task may not cancel the operation if it was already started on a worker thread.

Panics

awaiting the Task after the Executor is dropped will panic if the work was not already completed.

Examples

    let res = ex.spawn_blocking(move || {
        // Do some CPU-intensive or blocking work here.

        42
    }).await;

    assert_eq!(res, 42);

source

pub fn run(&self) -> Result<()>

Run the executor indefinitely, driving all spawned futures to completion. This method will block the current thread and only return in the case of an error.

Examples

      use std::thread;

      use cros_async::Executor;

      let ex = Executor::new();

      // Spawn a thread that runs the executor.
      let ex2 = ex.clone();
      thread::spawn(move || ex2.run());

      let task = ex.spawn(async { 7 + 13 });

      let result = ex.run_until(task)?;
      assert_eq!(result, 20);

source

pub fn run_until<F: Future>(&self, done: F) -> Result<F::Output>

Drive all futures spawned in this executor until f completes. This method will block the current thread only until f is complete and there may still be unfinished futures in the executor.

Examples

      use cros_async::Executor;

      let ex = Executor::new();

      let task = ex.spawn_local(async { 7 + 13 });

      let result = ex.run_until(task)?;
      assert_eq!(result, 20);