Merge pull request #71 from imdexsam/handle-reactive-checks-returning…

…-running

Handle reactive checks returning running

Cargo.toml

@@ -4,7 +4,7 @@ description = "Workflow framework based on the behavior trees"
 authors = ["BorisZhguchev <[email protected]>"]
 homepage = "https://github.com/besok/forester"
 repository = "https://github.com/besok/forester"
-version = "0.4.0"
+version = "0.4.1"
 edition = "2021"
 license-file = "LICENSE"
 

docs/src/falls.md

@@ -76,6 +76,8 @@ root main {
 The node `action` returns `running` and the whole sequence returns `running`
 but on the next tick it starts from the node `needs_to_charge` again.
 
-`r_fallback` will halt the `running` child to allow a graceful shutdown if a prior child changes from `failure` to `success`. In the above example, if `needs_to_change` returned `success` on the second tick then `action` would be halted before `r_fallback` returned `success` itself.
+`r_fallback` will halt the `running` child to allow a graceful shutdown if a prior child changes from `failure` to `success` or `running`. In the above example, if `needs_to_change` returned `success` on the second tick then `action` would be halted before `r_fallback` returned `success` itself.
+
+If `needs_to_charge` returned `running` on the second tick then `action` would also be halted, before potentially being restarted if `needs_to_charge` then returned `failure` again. Limiting the `r_fallback` node to a single `running` child avoids undefined behaviour in nodes that assume they are being run synchronously.
 
 Halting must be performed as quickly as possible. Note that currently only build-in flow, built-in decorator and sync action nodes are halted, async and remote actions are not.

docs/src/seq.md

@@ -98,6 +98,8 @@ root main {
 The node `perform_action` returns `running` and the whole sequence returns `running`
 but on the next tick it starts from the node `store` again.
 
-`r_sequence` will halt the `running` child to allow a graceful shutdown if a prior child changes from `success` to `failure`. In the above example, if `store` returned `failure` on the second tick then `perform_action` would be halted before `r_sequence` returned `failure` itself.
+`r_sequence` will halt the `running` child to allow a graceful shutdown if a prior child changes from `success` to `failure` or `running`. In the above example, if `store` returned `failure` on the second tick then `perform_action` would be halted before `r_sequence` returned `failure` itself.
+
+If `store` returned `running` on the second tick then `perform_action` would also be halted, before potentially being restarted if `store` then returned `success` again. Limiting the `r_sequence` node to a single `running` child avoids undefined behaviour in nodes that assume they are being run synchronously.
 
 Halting must be performed as quickly as possible. Note that currently only build-in flow, built-in decorator and sync action nodes are halted, async and remote actions are not.

src/runtime/forester.rs

@@ -230,15 +230,27 @@ impl Forester {
                                             debug!(target:"flow[run]", "tick:{}, {tpe}. stay with the new state: {}",ctx.curr_ts(),&ns);
                                             ctx.new_state(id, ns)?;
                                         }
-                                        FlowDecision::Halt(..) => {
-                                            return Err(RuntimeError::Unexpected("A child returning running should not trigger a flow node halt decision.".to_string()));
+                                        FlowDecision::Halt(new_state, halting_child_cursor) => {
+                                            // A reactively-checked child returning running will halt any other running child.
+                                            // Pop ourselves then halt the previously running child.
+                                            let halting_child_id = children[halting_child_cursor];
+                                            debug!(target:"flow[run]", "tick:{}, {tpe}. Reactively checked child '{child}' has returned running. Halting previously running child '{halting_child_id}', then go up with new state: {}.", ctx.curr_ts(), new_state);
+
+                                            ctx.new_state(id, new_state)?;
+                                            ctx.pop()?;
+
+                                            ctx.force_to_halting_state(halting_child_id)?;
+                                            ctx.push(halting_child_id)?;
                                         }
                                     }
                                 }
                             }
                             // Halting must be an atomic process, it is never spread over multiple ticks so should never be seen in this flow.
                             RNodeState::Halting(_) => {
-                                return Err(RuntimeError::Unexpected("A child returning running should not trigger a flow node halt decision.".to_string()));
+                                return Err(RuntimeError::Unexpected(
+                                    "A flow node child should never return a state of Halting."
+                                        .to_string(),
+                                ));
                             }
                             // child is finished, thus the node needs to make a decision how to proceed.
                             // this stage just updates the status and depending on the status,

src/runtime/forester/flow.rs

@@ -298,10 +298,29 @@ pub fn monitor(
 ) -> RtResult<FlowDecision> {
     match tpe {
         FlowType::RSequence | FlowType::RFallback => {
-            let cursor = read_cursor(tick_args.clone())?;
-            Ok(PopNode(RNodeState::Running(
-                tick_args.with(RUNNING_CHILD, RtValue::int(cursor)),
-            )))
+            // RSequence and RFallback don't use P_CURSOR
+            // let's get the cursor manually so P_CURSOR doesn't accidentially poison our result.
+            let cursor = tick_args
+                .find(CURSOR.to_string())
+                .and_then(RtValue::as_int)
+                .unwrap_or(0);
+            let previous_running_child = tick_args
+                .find(RUNNING_CHILD.to_string())
+                .and_then(RtValue::as_int);
+
+            let new_state =
+                RNodeState::Running(tick_args.with(RUNNING_CHILD, RtValue::int(cursor)));
+
+            // Check there isn't another child already being tracked by the RUNNING_CHILD key.
+            // If there is we'll need to halt that before going any further.
+            if let Some(prev_running_child_cursor) = previous_running_child {
+                if prev_running_child_cursor > cursor {
+                    return Ok(Halt(new_state, prev_running_child_cursor as usize));
+                }
+            }
+
+            // There was no other previously running child, continue as normal
+            Ok(PopNode(new_state))
         }
         FlowType::Sequence | FlowType::MSequence | FlowType::Fallback => {
             let cursor = read_cursor(tick_args.clone())?;

src/tests/flow.rs

@@ -322,6 +322,44 @@ fn r_sequence_halt_on_interrupt() {
     assert_eq!(x, 7)
 }
 
+#[test]
+fn r_sequence_halted_by_running() {
+    // See comment in the tree file for what this is testing.
+    let mut fb = fb("flow/r_sequence_halted_by_running");
+
+    fb.register_sync_action(
+        "incr",
+        GenerateData::new(|v| RtValue::int(v.as_int().unwrap_or(0) + 1)),
+    );
+
+    let mut f = fb.build().unwrap();
+    assert_eq!(f.run(), Ok(TickResult::success()));
+
+    let a =
+        f.bb.lock()
+            .unwrap()
+            .get("a".to_string())
+            .ok()
+            .flatten()
+            .unwrap()
+            .clone()
+            .as_int()
+            .unwrap();
+    assert_eq!(a, 10);
+
+    let b =
+        f.bb.lock()
+            .unwrap()
+            .get("b".to_string())
+            .ok()
+            .flatten()
+            .unwrap()
+            .clone()
+            .as_int()
+            .unwrap();
+    assert_eq!(b, 9);
+}
+
 #[test]
 fn fallback() {
     let mut fb = fb("flow/fallback");
@@ -489,6 +527,44 @@ fn r_fallback_halt_on_interrupt() {
     assert_eq!(x, 7)
 }
 
+#[test]
+fn r_fallback_halted_by_running() {
+    // See comment in the tree file for what this is testing.
+    let mut fb = fb("flow/r_fallback_halted_by_running");
+
+    fb.register_sync_action(
+        "incr",
+        GenerateData::new(|v| RtValue::int(v.as_int().unwrap_or(0) + 1)),
+    );
+
+    let mut f = fb.build().unwrap();
+    assert_eq!(f.run(), Ok(TickResult::success()));
+
+    let a =
+        f.bb.lock()
+            .unwrap()
+            .get("a".to_string())
+            .ok()
+            .flatten()
+            .unwrap()
+            .clone()
+            .as_int()
+            .unwrap();
+    assert_eq!(a, 10);
+
+    let b =
+        f.bb.lock()
+            .unwrap()
+            .get("b".to_string())
+            .ok()
+            .flatten()
+            .unwrap()
+            .clone()
+            .as_int()
+            .unwrap();
+    assert_eq!(b, 9);
+}
+
 #[test]
 fn parallel_simple() {
     turn_on_logs();

tree/tests/flow/r_fallback_halted_by_running/main.tree

@@ -0,0 +1,13 @@
+import "std::actions"
+impl incr(key:string,default:num);
+
+// We're testing that a reactively-checked child returning running will halt any other running children.
+// If a running child does halt other running children, repeat(5) will be reset and b will be incremented a total of 9 times. If no halt is called, b will only be incremented 5 times.
+root main repeat(2) r_fallback {
+    sequence {
+        incr("a",0)
+        equal(a,5)
+        repeat(2) success()
+    }
+    repeat(5) incr("b",0)
+}

tree/tests/flow/r_sequence_halted_by_running/main.tree

@@ -0,0 +1,13 @@
+import "std::actions"
+impl incr(key:string,default:num);
+
+// We're testing that a reactively-checked child returning running will halt any other running children.
+// If a running child does halt other running children, repeat(5) will be reset and b will be incremented a total of 9 times. If no halt is called, b will only be incremented 5 times.
+root main retry(2) r_sequence {
+    fallback {
+        inverter incr("a",0)
+        inverter equal(a,5)
+        retry(2) fail_empty()
+    }
+    repeat(5) incr("b",0)
+}