Struct Executor 

pub struct Executor<S>
where
    S: Strategy,
{ /* private fields */ }

Executor.

Executors serve as the primary interface for submitting and monitoring tasks within the system. They act as a frontend to various execution Strategy implementations, which define how tasks are prioritized and executed. Each execution Strategy encapsulates an implementation that determines the order and concurrency of execution. Abstracting the execution mechanism allows for flexible and interchangeable task management strategies.

Additionally, executors implement Clone, which allows to easily share them among different parts of the system without borrowing issues.

Note that executors are not responsible for managing the lifetime of tasks, as it is assumed that tasks are self-contained and can be run independently. If a Task is submitted to an executor, it can’t be cancelled or stopped, as the executor is not aware of the task’s internal state. However, callers can implement fine-grained execution strategies on top of the executor to gain fine-grained control over task execution.

This is an opinionated implementation that specifically targets the needs of our execution model. It is not meant to be a general-purpose executor.

Examples

Create an executor spawning 8 tasks using all CPUs - 1:

use std::thread;
use std::time::Duration;
use zrx_executor::Executor;

// Create executor
let executor = Executor::default();

// Create tasks up to the executor's capacity
for _ in 0..executor.capacity() {
    executor.submit(|| {
        thread::sleep(Duration::from_millis(20));
    })?;
}

Implementations

impl<S> Executor<S>

where S: Strategy,

pub fn new(strategy: S) -> Self

Creates an executor.

Examples

use zrx_executor::strategy::WorkSharing;
use zrx_executor::Executor;

// Create executor with strategy
let executor = Executor::new(WorkSharing::default());

pub fn submit<T>(&self, task: T) -> Result

where T: Into<Box<dyn Task>>,

Submits a task.

This method submits a Task, which is executed by one of the worker threads as soon as possible. If a task computes a result, a Sender can be shared with the task, to send the result back to the caller, which can then poll a Receiver.

Note that tasks are intended to only run once, which is why they are consumed. If a task needs to be run multiple times, it must be wrapped in a closure that creates a new task each time. This allows for safe sharing of state between tasks.

Errors

If the executor encounters a problem during task submission, it will forward the encountered error to the caller, returning the task. Most likely, the underlying execution strategy is at capacity, which means the caller should resubmit the task at a later time. This is possible, since this method accepts any type that implements the Task trait and converts it into a boxed task.

Examples

Submit a task:

use zrx_executor::Executor;

// Create executor and submit task
let executor = Executor::default();
executor.submit(|| println!("Task"))?;

Submit a task returning subtasks:

use zrx_executor::Executor;

// Create executor and submit task
let executor = Executor::default();
executor.submit(|| {
    println!("Task 1");
    || {
        println!("Task 1.1");
        || {
            println!("Task 1.1.1");
        }
    }
})?;

Submit a task returning a task collection:

use zrx_executor::task::Tasks;
use zrx_executor::Executor;

// Create executor and submit task
let executor = Executor::default();
executor.submit(|| {
    println!("Task 1");

    // Create subtasks
    let mut tasks = Tasks::new();
    tasks.add(|| println!("Task 1.1"));
    tasks.add(|| println!("Task 1.2"));
    tasks.add(|| println!("Task 1.3"));
    tasks
})?;

pub fn wait(&self)

Waits for all tasks to finish.

This method blocks the current thread until all submitted running and pending tasks have been completed. Calling this method is not necessary, as it’s called automatically when the executor is dropped, but it might be helpful for testing and debugging purposes.

Examples

use std::thread;
use std::time::Duration;
use zrx_executor::Executor;

// Create executor
let executor = Executor::default();

// Create tasks up to the executor's capacity
for _ in 0..executor.capacity() {
    executor.submit(|| {
        thread::sleep(Duration::from_millis(20));
    })?;
}

// Wait for all tasks to finish
executor.wait();
assert!(executor.is_empty());

impl<S> Executor<S>

where S: Strategy,

pub fn len(&self) -> usize

Returns the number of tasks.

This method returns the total number of tasks currently managed by the executor, which includes running as well as pending tasks.

Examples

use zrx_executor::Executor;

// Get number of tasks
let executor = Executor::default();
assert_eq!(executor.len(), 0);

pub fn is_empty(&self) -> bool

Returns whether there are any tasks.

This method checks whether the executor has running or pending tasks, and if not, considers the executor as idle. It’s particularly useful for waiting until an executor has processed all tasks, which is necessary for implementing schedulers on top of executors.

Examples

use zrx_executor::Executor;

// Check whether executor is idle
let executor = Executor::default();
assert!(executor.is_empty());

pub fn is_saturated(&self) -> bool

Returns whether the executor is saturated.

This method checks whether the executor is at capacity, which means task submission will fail until a worker has finished a task.

Examples

use zrx_executor::Executor;

// Check whether executor is saturated
let executor = Executor::default();
assert!(!executor.is_saturated());

pub fn num_workers(&self) -> usize

Returns the number of workers.

Examples

use zrx_executor::strategy::WorkSharing;
use zrx_executor::Executor;

// Get number of workers
let executor = Executor::new(WorkSharing::new(1));
assert_eq!(executor.num_workers(), 1);

pub fn num_tasks_running(&self) -> usize

Returns the number of running tasks.

This method allows to monitor the worker load, as it returns how many workers are currently actively executing tasks.

Examples

use zrx_executor::Executor;

// Get number of running tasks
let executor = Executor::default();
assert_eq!(executor.num_tasks_running(), 0);

pub fn num_tasks_pending(&self) -> usize

Returns the number of pending tasks.

This method allows to throttle the submission of tasks, as it returns how many tasks are currently waiting to be executed.

Examples

use zrx_executor::Executor;

// Get number of pending tasks
let executor = Executor::default();
assert_eq!(executor.num_tasks_pending(), 0);

pub fn capacity(&self) -> usize

Returns the capacity.

This method returns the maximum number of tasks that can be submitted at once, which can be used by the strategy for applying backpressure.

Examples

use zrx_executor::Executor;

// Get maximum number of tasks
let executor = Executor::default();
assert!(executor.capacity() >= executor.num_workers());

Trait Implementations

impl<S> Clone for Executor<S>

where S: Strategy,

fn clone(&self) -> Self

Clones the executor.

This method creates a new executor with the same execution strategy, which allows to share them without borrowing issues.

Examples

use zrx_executor::Executor;

// Create and clone executor
let executor = Executor::default();
executor.clone();

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<S> Debug for Executor<S>

where S: Strategy + Debug,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Executor<WorkSharing>

fn default() -> Self

Creates an executor using the default work-sharing strategy.

Examples

use zrx_executor::Executor;

// Create executor
let executor = Executor::default();

impl<S> Drop for Executor<S>

where S: Strategy,

fn drop(&mut self)

Waits for all tasks to finish.