Struct ParallelExecutor 

pub struct ParallelExecutor { /* private fields */ }

Parallel executor that identifies independent nodes and executes them concurrently

Implementations

impl ParallelExecutor

pub fn new(config: ParallelConfig) -> Self

Create a new parallel executor

Examples found in repository

examples/parallel_execution.rs (lines 292-295)

async fn analyze_and_execute(def: GraphDef) -> anyhow::Result<()> {
    // Create parallel executor
    let mut executor = ParallelExecutor::new(ParallelConfig {
        max_concurrent: 10,
        verbose: true,
    });

    // Register nodes
    for node_id in def.nodes.keys() {
        let node: Box<dyn Node> = Box::new(RuleNode::new(node_id, "true"));
        executor.register_node(node);
    }

    // Analyze parallelism
    let stats = executor.get_parallelism_stats(&def)?;
    stats.print_summary();

    // Execute the graph
    let mut graph = Graph::new(def);
    let start = Instant::now();
    executor.execute(&mut graph).await?;
    let duration = start.elapsed();

    println!("Execution completed in {:?}", duration);

    Ok(())
}

pub fn register_node(&mut self, node: Box<dyn Node>)

Register a node with the executor

Examples found in repository

examples/parallel_execution.rs (line 300)

async fn analyze_and_execute(def: GraphDef) -> anyhow::Result<()> {
    // Create parallel executor
    let mut executor = ParallelExecutor::new(ParallelConfig {
        max_concurrent: 10,
        verbose: true,
    });

    // Register nodes
    for node_id in def.nodes.keys() {
        let node: Box<dyn Node> = Box::new(RuleNode::new(node_id, "true"));
        executor.register_node(node);
    }

    // Analyze parallelism
    let stats = executor.get_parallelism_stats(&def)?;
    stats.print_summary();

    // Execute the graph
    let mut graph = Graph::new(def);
    let start = Instant::now();
    executor.execute(&mut graph).await?;
    let duration = start.elapsed();

    println!("Execution completed in {:?}", duration);

    Ok(())
}

pub fn identify_layers(&self, def: &GraphDef) -> Result<Vec<ExecutionLayer>>

Analyze graph and identify execution layers Nodes in the same layer have no dependencies on each other and can run in parallel

pub async fn execute(&self, graph: &mut Graph) -> Result<()>

Execute the entire graph with parallel execution per layer

Examples found in repository

examples/parallel_execution.rs (line 310)

async fn analyze_and_execute(def: GraphDef) -> anyhow::Result<()> {
    // Create parallel executor
    let mut executor = ParallelExecutor::new(ParallelConfig {
        max_concurrent: 10,
        verbose: true,
    });

    // Register nodes
    for node_id in def.nodes.keys() {
        let node: Box<dyn Node> = Box::new(RuleNode::new(node_id, "true"));
        executor.register_node(node);
    }

    // Analyze parallelism
    let stats = executor.get_parallelism_stats(&def)?;
    stats.print_summary();

    // Execute the graph
    let mut graph = Graph::new(def);
    let start = Instant::now();
    executor.execute(&mut graph).await?;
    let duration = start.elapsed();

    println!("Execution completed in {:?}", duration);

    Ok(())
}

pub fn get_parallelism_stats(&self, def: &GraphDef) -> Result<ParallelismStats>

Get parallel execution statistics

Examples found in repository

examples/parallel_execution.rs (line 304)

async fn analyze_and_execute(def: GraphDef) -> anyhow::Result<()> {
    // Create parallel executor
    let mut executor = ParallelExecutor::new(ParallelConfig {
        max_concurrent: 10,
        verbose: true,
    });

    // Register nodes
    for node_id in def.nodes.keys() {
        let node: Box<dyn Node> = Box::new(RuleNode::new(node_id, "true"));
        executor.register_node(node);
    }

    // Analyze parallelism
    let stats = executor.get_parallelism_stats(&def)?;
    stats.print_summary();

    // Execute the graph
    let mut graph = Graph::new(def);
    let start = Instant::now();
    executor.execute(&mut graph).await?;
    let duration = start.elapsed();

    println!("Execution completed in {:?}", duration);

    Ok(())
}

/// Compare sequential vs parallel execution performance
async fn compare_performance() -> anyhow::Result<()> {
    println!("Comparing sequential vs parallel execution...\n");

    // Create a graph with good parallelism
    let def = create_wide_graph();

    // Sequential execution simulation
    let sequential_time = def.nodes.len(); // Assume 1 unit per node
    println!("Sequential execution time (simulated): {} units", sequential_time);

    // Parallel execution analysis
    let executor = ParallelExecutor::default();
    let stats = executor.get_parallelism_stats(&def)?;
    let parallel_time = stats.num_layers; // Each layer is 1 unit

    println!("Parallel execution time (simulated): {} units", parallel_time);
    println!("Speedup: {:.2}x", sequential_time as f64 / parallel_time as f64);
    println!("Efficiency: {:.1}%", (stats.theoretical_speedup / def.nodes.len() as f64) * 100.0);

    Ok(())
}

Trait Implementations

impl Default for ParallelExecutor

fn default() -> Self

Returns the “default value” for a type.