Run parallel steps concurrently

Author: njfio

README.md

@@ -238,7 +238,7 @@ This will execute the `my_pipeline` pipeline with the initial input "hello world
 * **Variable Substitution:** Pipeline steps can use variables defined in the state using `${variable_name}` syntax.
 * **Retry Mechanism:** Steps can be configured with a retry mechanism to handle transient errors.
 * **State Persistence:** Pipeline state is automatically saved and loaded, allowing for seamless resumption.
-* **Parallel Execution:** The `Parallel` step allows for concurrent execution of steps.
+* **Parallel Execution:** The `Parallel` step allows for concurrent execution of steps. Nested `Parallel` blocks are also run concurrently and their results aggregated when all tasks finish.
 * **Timeout Mechanism:** The `Timeout` step allows for setting a time limit for a specific step.
 
 ### Example Pipeline

crates/fluent-engines/src/pipeline_executor.rs

@@ -723,13 +723,41 @@ impl<S: StateStore + Clone + std::marker::Sync + std::marker::Send> PipelineExec
                     }
                 }
                 PipelineStep::Parallel { name: _, steps } => {
-                    // For simplicity, we'll execute parallel steps sequentially in this context
-                    let mut result = HashMap::new();
-                    for sub_step in steps {
-                        let step_result = Self::execute_single_step(sub_step, state).await?;
-                        result.extend(step_result);
+                    let state_arc = Arc::new(tokio::sync::Mutex::new(state.clone()));
+                    let mut set = JoinSet::new();
+
+                    for sub_step in steps.iter().cloned() {
+                        let state_clone = Arc::clone(&state_arc);
+                        set.spawn(async move {
+                            let mut guard = state_clone.lock().await;
+                            Self::execute_single_step(&sub_step, &mut guard).await
+                        });
                     }
-                    Ok(result)
+
+                    let mut combined_results = HashMap::new();
+                    while let Some(result) = set.join_next().await {
+                        match result {
+                            Ok(Ok(step_result)) => {
+                                combined_results.extend(step_result);
+                            }
+                            Ok(Err(e)) => {
+                                combined_results
+                                    .insert(format!("error_{}", combined_results.len()), e.to_string());
+                            }
+                            Err(e) => {
+                                combined_results.insert(
+                                    format!("join_error_{}", combined_results.len()),
+                                    e.to_string(),
+                                );
+                            }
+                        }
+                    }
+
+                    let state_guard = state_arc.lock().await;
+                    state.data.extend(state_guard.data.clone());
+                    state.data.extend(combined_results.clone());
+
+                    Ok(combined_results)
                 }
                 _ => Err(anyhow!("Unknown step type")),
             }