KCL: Compute pipe expressions with loops, not recursion (#1941)

On `main`, the new test program in `tests/executor` causes a stack overflow. Running a flamegraph via `sudo cargo flamegraph --test executor -- serial_test_mike_stress_lines` shows that the problem: `fn execute_pipe_body` is very recursive. 

When there's a long pipe, `execute_pipe_body` executes the next child expression, which is always a CallExpression. So `execute_pipe_body` calls `CallExpression::execute`. Which then calls `execute_pipe_body` again. 

Fix is simple: `execute_pipe_body` now iterates over pipe subexpressions instead of recursing. This fixes the stack overflow, and is much faster too.

Closes https://github.com/KittyCAD/modeling-app/issues/1891.

src/wasm-lib/kcl/src/ast/types.rs

@@ -715,15 +715,9 @@ impl BinaryPart {
     pub async fn get_result(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
-        // We DO NOT set this globally because if we did and this was called inside a pipe it would
-        // stop the execution of the pipe.
-        // THIS IS IMPORTANT.
-        let mut new_pipe_info = pipe_info.clone();
-        new_pipe_info.is_in_pipe = false;
-
         match self {
             BinaryPart::Literal(literal) => Ok(literal.into()),
             BinaryPart::Identifier(identifier) => {
@@ -731,14 +725,10 @@ impl BinaryPart {
                 Ok(value.clone())
             }
             BinaryPart::BinaryExpression(binary_expression) => {
-                binary_expression.get_result(memory, &mut new_pipe_info, ctx).await
-            }
-            BinaryPart::CallExpression(call_expression) => {
-                call_expression.execute(memory, &mut new_pipe_info, ctx).await
-            }
-            BinaryPart::UnaryExpression(unary_expression) => {
-                unary_expression.get_result(memory, &mut new_pipe_info, ctx).await
+                binary_expression.get_result(memory, pipe_info, ctx).await
             }
+            BinaryPart::CallExpression(call_expression) => call_expression.execute(memory, pipe_info, ctx).await,
+            BinaryPart::UnaryExpression(unary_expression) => unary_expression.get_result(memory, pipe_info, ctx).await,
             BinaryPart::MemberExpression(member_expression) => member_expression.get_result(memory),
         }
     }
@@ -1019,7 +1009,7 @@ impl CallExpression {
     pub async fn execute(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
         let fn_name = self.callee.name.clone();
@@ -1037,14 +1027,7 @@ impl CallExpression {
                 Value::BinaryExpression(binary_expression) => {
                     binary_expression.get_result(memory, pipe_info, ctx).await?
                 }
-                Value::CallExpression(call_expression) => {
-                    // We DO NOT set this globally because if we did and this was called inside a pipe it would
-                    // stop the execution of the pipe.
-                    // THIS IS IMPORTANT.
-                    let mut new_pipe_info = pipe_info.clone();
-                    new_pipe_info.is_in_pipe = false;
-                    call_expression.execute(memory, &mut new_pipe_info, ctx).await?
-                }
+                Value::CallExpression(call_expression) => call_expression.execute(memory, pipe_info, ctx).await?,
                 Value::UnaryExpression(unary_expression) => unary_expression.get_result(memory, pipe_info, ctx).await?,
                 Value::ObjectExpression(object_expression) => object_expression.execute(memory, pipe_info, ctx).await?,
                 Value::ArrayExpression(array_expression) => array_expression.execute(memory, pipe_info, ctx).await?,
@@ -1056,7 +1039,7 @@ impl CallExpression {
                 }
                 Value::PipeSubstitution(pipe_substitution) => pipe_info
                     .previous_results
-                    .get(&pipe_info.index - 1)
+                    .as_ref()
                     .ok_or_else(|| {
                         KclError::Semantic(KclErrorDetails {
                             message: format!("PipeSubstitution index out of bounds: {:?}", pipe_info),
@@ -1081,14 +1064,8 @@ impl CallExpression {
                 // Attempt to call the function.
                 let args = crate::std::Args::new(fn_args, self.into(), ctx.clone());
                 let result = func.std_lib_fn()(args).await?;
-                if pipe_info.is_in_pipe {
-                    pipe_info.index += 1;
-                    pipe_info.previous_results.push(result);
-                    execute_pipe_body(memory, &pipe_info.body.clone(), pipe_info, self.into(), ctx).await
-                } else {
                 Ok(result)
             }
-            }
             FunctionKind::Std(func) => {
                 let function_expression = func.function();
                 let parts = function_expression.clone().into_parts().map_err(|e| {
@@ -1152,15 +1129,8 @@ impl CallExpression {
                 })?;
                 let result = result.get_value()?;
 
-                if pipe_info.is_in_pipe {
-                    pipe_info.index += 1;
-                    pipe_info.previous_results.push(result);
-
-                    execute_pipe_body(memory, &pipe_info.body.clone(), pipe_info, self.into(), ctx).await
-                } else {
                 Ok(result)
             }
-            }
             FunctionKind::UserDefined => {
                 let func = memory.get(&fn_name, self.into())?;
                 let result = func
@@ -1175,17 +1145,10 @@ impl CallExpression {
 
                 let result = result.get_value()?;
 
-                if pipe_info.is_in_pipe {
-                    pipe_info.index += 1;
-                    pipe_info.previous_results.push(result);
-
-                    execute_pipe_body(memory, &pipe_info.body.clone(), pipe_info, self.into(), ctx).await
-                } else {
                 Ok(result)
             }
         }
     }
-    }
 
     /// Returns a hover value that includes the given character position.
     pub fn get_hover_value_for_position(&self, pos: usize, code: &str) -> Option<Hover> {
@@ -1731,7 +1694,7 @@ impl ArrayExpression {
     pub async fn execute(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
         let mut results = Vec::with_capacity(self.elements.len());
@@ -1747,14 +1710,7 @@ impl ArrayExpression {
                 Value::BinaryExpression(binary_expression) => {
                     binary_expression.get_result(memory, pipe_info, ctx).await?
                 }
-                Value::CallExpression(call_expression) => {
-                    // We DO NOT set this globally because if we did and this was called inside a pipe it would
-                    // stop the execution of the pipe.
-                    // THIS IS IMPORTANT.
-                    let mut new_pipe_info = pipe_info.clone();
-                    new_pipe_info.is_in_pipe = false;
-                    call_expression.execute(memory, &mut new_pipe_info, ctx).await?
-                }
+                Value::CallExpression(call_expression) => call_expression.execute(memory, pipe_info, ctx).await?,
                 Value::UnaryExpression(unary_expression) => unary_expression.get_result(memory, pipe_info, ctx).await?,
                 Value::ObjectExpression(object_expression) => object_expression.execute(memory, pipe_info, ctx).await?,
                 Value::ArrayExpression(array_expression) => array_expression.execute(memory, pipe_info, ctx).await?,
@@ -1885,7 +1841,7 @@ impl ObjectExpression {
     pub async fn execute(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
         let mut object = Map::new();
@@ -1900,14 +1856,7 @@ impl ObjectExpression {
                 Value::BinaryExpression(binary_expression) => {
                     binary_expression.get_result(memory, pipe_info, ctx).await?
                 }
-                Value::CallExpression(call_expression) => {
-                    // We DO NOT set this globally because if we did and this was called inside a pipe it would
-                    // stop the execution of the pipe.
-                    // THIS IS IMPORTANT.
-                    let mut new_pipe_info = pipe_info.clone();
-                    new_pipe_info.is_in_pipe = false;
-                    call_expression.execute(memory, &mut new_pipe_info, ctx).await?
-                }
+                Value::CallExpression(call_expression) => call_expression.execute(memory, pipe_info, ctx).await?,
                 Value::UnaryExpression(unary_expression) => unary_expression.get_result(memory, pipe_info, ctx).await?,
                 Value::ObjectExpression(object_expression) => object_expression.execute(memory, pipe_info, ctx).await?,
                 Value::ArrayExpression(array_expression) => array_expression.execute(memory, pipe_info, ctx).await?,
@@ -2338,25 +2287,11 @@ impl BinaryExpression {
     pub async fn get_result(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
-        // We DO NOT set this globally because if we did and this was called inside a pipe it would
-        // stop the execution of the pipe.
-        // THIS IS IMPORTANT.
-        let mut new_pipe_info = pipe_info.clone();
-        new_pipe_info.is_in_pipe = false;
-
-        let left_json_value = self
-            .left
-            .get_result(memory, &mut new_pipe_info, ctx)
-            .await?
-            .get_json_value()?;
-        let right_json_value = self
-            .right
-            .get_result(memory, &mut new_pipe_info, ctx)
-            .await?
-            .get_json_value()?;
+        let left_json_value = self.left.get_result(memory, pipe_info, ctx).await?.get_json_value()?;
+        let right_json_value = self.right.get_result(memory, pipe_info, ctx).await?.get_json_value()?;
 
         // First check if we are doing string concatenation.
         if self.operator == BinaryOperator::Add {
@@ -2542,19 +2477,13 @@ impl UnaryExpression {
     pub async fn get_result(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
-        // We DO NOT set this globally because if we did and this was called inside a pipe it would
-        // stop the execution of the pipe.
-        // THIS IS IMPORTANT.
-        let mut new_pipe_info = pipe_info.clone();
-        new_pipe_info.is_in_pipe = false;
-
         let num = parse_json_number_as_f64(
             &self
                 .argument
-                .get_result(memory, &mut new_pipe_info, ctx)
+                .get_result(memory, pipe_info, ctx)
                 .await?
                 .get_json_value()?,
             self.into(),
@@ -2606,6 +2535,8 @@ pub enum UnaryOperator {
 pub struct PipeExpression {
     pub start: usize,
     pub end: usize,
+    // TODO: Only the first body expression can be any Value.
+    // The rest will be CallExpression, and the AST type should reflect this.
     pub body: Vec<Value>,
     pub non_code_meta: NonCodeMeta,
 }
@@ -2696,12 +2627,9 @@ impl PipeExpression {
     pub async fn get_result(
         &self,
         memory: &mut ProgramMemory,
-        pipe_info: &mut PipeInfo,
+        pipe_info: &PipeInfo,
         ctx: &ExecutorContext,
     ) -> Result<MemoryItem, KclError> {
-        // Reset the previous results.
-        pipe_info.previous_results = vec![];
-        pipe_info.index = 0;
         execute_pipe_body(memory, &self.body, pipe_info, self.into(), ctx).await
     }
 
@@ -2717,57 +2645,59 @@ impl PipeExpression {
 async fn execute_pipe_body(
     memory: &mut ProgramMemory,
     body: &[Value],
-    pipe_info: &mut PipeInfo,
+    pipe_info: &PipeInfo,
     source_range: SourceRange,
     ctx: &ExecutorContext,
 ) -> Result<MemoryItem, KclError> {
-    if pipe_info.index == body.len() {
-        pipe_info.is_in_pipe = false;
-        return Ok(pipe_info
-            .previous_results
-            .last()
-            .ok_or_else(|| {
+    let mut body_iter = body.iter();
+    let first = body_iter.next().ok_or_else(|| {
         KclError::Semantic(KclErrorDetails {
-                    message: "Pipe body results should have at least one expression".to_string(),
-                    source_ranges: vec![source_range],
-                })
-            })?
-            .clone());
-    }
-
-    let expression = body.get(pipe_info.index).ok_or_else(|| {
-        KclError::Semantic(KclErrorDetails {
-            message: format!("Invalid index for pipe: {}", pipe_info.index),
+            message: "Pipe expressions cannot be empty".to_owned(),
             source_ranges: vec![source_range],
         })
     })?;
-
-    match expression {
-        Value::BinaryExpression(binary_expression) => {
-            let result = binary_expression.get_result(memory, pipe_info, ctx).await?;
-            pipe_info.previous_results.push(result);
-            pipe_info.index += 1;
-            execute_pipe_body(memory, body, pipe_info, source_range, ctx).await
-        }
-        Value::CallExpression(call_expression) => {
-            pipe_info.is_in_pipe = true;
-            pipe_info.body = body.to_vec();
-            call_expression.execute(memory, pipe_info, ctx).await
-        }
-        Value::Identifier(identifier) => {
-            let result = memory.get(&identifier.name, identifier.into())?;
-            pipe_info.previous_results.push(result.clone());
-            pipe_info.index += 1;
-            execute_pipe_body(memory, body, pipe_info, source_range, ctx).await
-        }
+    // Evaluate the first element in the pipeline.
+    // They use the `pipe_info` from some AST node above this, so that if pipe expression is nested in a larger pipe expression,
+    // they use the % from the parent. After all, this pipe expression hasn't been executed yet, so it doesn't have any % value
+    // of its own.
+    let output = match first {
+        Value::BinaryExpression(binary_expression) => binary_expression.get_result(memory, pipe_info, ctx).await?,
+        Value::CallExpression(call_expression) => call_expression.execute(memory, pipe_info, ctx).await?,
+        Value::Identifier(identifier) => memory.get(&identifier.name, identifier.into())?.clone(),
         _ => {
             // Return an error this should not happen.
-            Err(KclError::Semantic(KclErrorDetails {
+            return Err(KclError::Semantic(KclErrorDetails {
+                message: format!("PipeExpression not implemented here: {:?}", first),
+                source_ranges: vec![first.into()],
+            }));
+        }
+    };
+    // Now that we've evaluated the first child expression in the pipeline, following child expressions
+    // should use the previous child expression for %.
+    // This means there's no more need for the previous `pipe_info` from the parent AST node above this one.
+    let mut new_pipe_info = PipeInfo::new();
+    new_pipe_info.previous_results = Some(output);
+    // Evaluate remaining elements.
+    for expression in body {
+        let output = match expression {
+            Value::BinaryExpression(binary_expression) => {
+                binary_expression.get_result(memory, &new_pipe_info, ctx).await?
+            }
+            Value::CallExpression(call_expression) => call_expression.execute(memory, &new_pipe_info, ctx).await?,
+            Value::Identifier(identifier) => memory.get(&identifier.name, identifier.into())?.clone(),
+            _ => {
+                // Return an error this should not happen.
+                return Err(KclError::Semantic(KclErrorDetails {
                     message: format!("PipeExpression not implemented here: {:?}", expression),
                     source_ranges: vec![expression.into()],
-            }))
+                }));
             }
+        };
+        new_pipe_info.previous_results = Some(output);
     }
+    // Safe to unwrap here, because `newpipe_info` always has something pushed in when the `match first` executes.
+    let final_output = new_pipe_info.previous_results.unwrap();
+    Ok(final_output)
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, Bake, FromStr, Display)]

src/wasm-lib/kcl/src/executor.rs

@@ -953,20 +953,12 @@ impl ExtrudeSurface {
 #[ts(export)]
 #[serde(rename_all = "camelCase")]
 pub struct PipeInfo {
-    pub previous_results: Vec<MemoryItem>,
-    pub is_in_pipe: bool,
-    pub index: usize,
-    pub body: Vec<Value>,
+    pub previous_results: Option<MemoryItem>,
 }
 
 impl PipeInfo {
     pub fn new() -> Self {
-        Self {
-            previous_results: Vec::new(),
-            is_in_pipe: false,
-            index: 0,
-            body: Vec::new(),
-        }
+        Self { previous_results: None }
     }
 }
 
@@ -1022,14 +1014,14 @@ pub async fn execute(
         )
         .await?;
 
-    let mut pipe_info = PipeInfo::default();
+    let pipe_info = PipeInfo::default();
 
     // Iterate over the body of the program.
     for statement in &program.body {
         match statement {
             BodyItem::ExpressionStatement(expression_statement) => {
                 if let Value::PipeExpression(pipe_expr) = &expression_statement.expression {
-                    pipe_expr.get_result(memory, &mut pipe_info, ctx).await?;
+                    pipe_expr.get_result(memory, &pipe_info, ctx).await?;
                 } else if let Value::CallExpression(call_expr) = &expression_statement.expression {
                     let fn_name = call_expr.callee.name.to_string();
                     let mut args: Vec<MemoryItem> = Vec::new();
@@ -1041,23 +1033,23 @@ pub async fn execute(
                                 args.push(memory_item.clone());
                             }
                             Value::CallExpression(call_expr) => {
-                                let result = call_expr.execute(memory, &mut pipe_info, ctx).await?;
+                                let result = call_expr.execute(memory, &pipe_info, ctx).await?;
                                 args.push(result);
                             }
                             Value::BinaryExpression(binary_expression) => {
-                                let result = binary_expression.get_result(memory, &mut pipe_info, ctx).await?;
+                                let result = binary_expression.get_result(memory, &pipe_info, ctx).await?;
                                 args.push(result);
                             }
                             Value::UnaryExpression(unary_expression) => {
-                                let result = unary_expression.get_result(memory, &mut pipe_info, ctx).await?;
+                                let result = unary_expression.get_result(memory, &pipe_info, ctx).await?;
                                 args.push(result);
                             }
                             Value::ObjectExpression(object_expression) => {
-                                let result = object_expression.execute(memory, &mut pipe_info, ctx).await?;
+                                let result = object_expression.execute(memory, &pipe_info, ctx).await?;
                                 args.push(result);
                             }
                             Value::ArrayExpression(array_expression) => {
-                                let result = array_expression.execute(memory, &mut pipe_info, ctx).await?;
+                                let result = array_expression.execute(memory, &pipe_info, ctx).await?;
                                 args.push(result);
                             }
                             // We do nothing for the rest.
@@ -1108,7 +1100,7 @@ pub async fn execute(
                             memory.add(&var_name, value.clone(), source_range)?;
                         }
                         Value::BinaryExpression(binary_expression) => {
-                            let result = binary_expression.get_result(memory, &mut pipe_info, ctx).await?;
+                            let result = binary_expression.get_result(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::FunctionExpression(function_expression) => {
@@ -1145,11 +1137,11 @@ pub async fn execute(
                             )?;
                         }
                         Value::CallExpression(call_expression) => {
-                            let result = call_expression.execute(memory, &mut pipe_info, ctx).await?;
+                            let result = call_expression.execute(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::PipeExpression(pipe_expression) => {
-                            let result = pipe_expression.get_result(memory, &mut pipe_info, ctx).await?;
+                            let result = pipe_expression.get_result(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::PipeSubstitution(pipe_substitution) => {
@@ -1162,11 +1154,11 @@ pub async fn execute(
                             }));
                         }
                         Value::ArrayExpression(array_expression) => {
-                            let result = array_expression.execute(memory, &mut pipe_info, ctx).await?;
+                            let result = array_expression.execute(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::ObjectExpression(object_expression) => {
-                            let result = object_expression.execute(memory, &mut pipe_info, ctx).await?;
+                            let result = object_expression.execute(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::MemberExpression(member_expression) => {
@@ -1174,7 +1166,7 @@ pub async fn execute(
                             memory.add(&var_name, result, source_range)?;
                         }
                         Value::UnaryExpression(unary_expression) => {
-                            let result = unary_expression.get_result(memory, &mut pipe_info, ctx).await?;
+                            let result = unary_expression.get_result(memory, &pipe_info, ctx).await?;
                             memory.add(&var_name, result, source_range)?;
                         }
                     }
@@ -1182,11 +1174,11 @@ pub async fn execute(
             }
             BodyItem::ReturnStatement(return_statement) => match &return_statement.argument {
                 Value::BinaryExpression(bin_expr) => {
-                    let result = bin_expr.get_result(memory, &mut pipe_info, ctx).await?;
+                    let result = bin_expr.get_result(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::UnaryExpression(unary_expr) => {
-                    let result = unary_expr.get_result(memory, &mut pipe_info, ctx).await?;
+                    let result = unary_expr.get_result(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::Identifier(identifier) => {
@@ -1197,15 +1189,15 @@ pub async fn execute(
                     memory.return_ = Some(ProgramReturn::Value(literal.into()));
                 }
                 Value::ArrayExpression(array_expr) => {
-                    let result = array_expr.execute(memory, &mut pipe_info, ctx).await?;
+                    let result = array_expr.execute(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::ObjectExpression(obj_expr) => {
-                    let result = obj_expr.execute(memory, &mut pipe_info, ctx).await?;
+                    let result = obj_expr.execute(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::CallExpression(call_expr) => {
-                    let result = call_expr.execute(memory, &mut pipe_info, ctx).await?;
+                    let result = call_expr.execute(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::MemberExpression(member_expr) => {
@@ -1213,7 +1205,7 @@ pub async fn execute(
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::PipeExpression(pipe_expr) => {
-                    let result = pipe_expr.get_result(memory, &mut pipe_info, ctx).await?;
+                    let result = pipe_expr.get_result(memory, &pipe_info, ctx).await?;
                     memory.return_ = Some(ProgramReturn::Value(result));
                 }
                 Value::PipeSubstitution(_) => {}

src/wasm-lib/tests/executor/main.rs

@@ -142,6 +142,15 @@ const part002 = startSketchOn(part001, "start")
     twenty_twenty::assert_image("tests/executor/outputs/sketch_on_face_start.png", &result, 0.999);
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn serial_test_mike_stress_lines() {
+    let code = include_str!("inputs/mike_stress_test.kcl");
+    let result = execute_and_snapshot(code, kittycad::types::UnitLength::Mm)
+        .await
+        .unwrap();
+    twenty_twenty::assert_image("tests/executor/outputs/mike_stress_test.png", &result, 0.999);
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn serial_test_sketch_on_face_end() {
     let code = r#"fn cube = (pos, scale) => {
@@ -493,7 +502,7 @@ async fn serial_test_execute_i_shape() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-#[ignore] // ignore until more stack fixes
+#[ignore] // No longer a stack overflow problem, instead it causes an engine internal error.
 async fn serial_test_execute_pipes_on_pipes() {
     let code = include_str!("inputs/pipes_on_pipes.kcl");
 
@@ -514,7 +523,6 @@ async fn serial_test_execute_cylinder() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-#[ignore = "currently stack overflows"]
 async fn serial_test_execute_kittycad_svg() {
     let code = include_str!("inputs/kittycad_svg.kcl");