Debug stepping feature (#178)

Signed-off-by: Michael X. Grey <greyxmike@gmail.com>

Author: mxgrey

src/debug.rs

@@ -38,6 +38,9 @@ use crate::{
 #[cfg(feature = "single_threaded_async")]
 use crate::async_execution::SingleThreadedExecution;
 
+#[cfg(feature = "trace")]
+use crate::{Debug, DebugRoster};
+
 #[derive(Resource, Default, Clone, Copy)]
 pub struct FlushParameters {
     /// By default, a flush will loop until the whole [`OperationRoster`] is empty.
@@ -87,11 +90,29 @@ pub fn flush_execution() -> ScheduleConfigs<ScheduleSystem> {
 fn flush_execution_impl(
     world: &mut World,
     new_service_query: &mut QueryState<(Entity, &mut ServiceHook), Added<ServiceHook>>,
+    #[cfg(feature = "trace")] debug: &mut SystemState<Option<Res<Debug>>>,
 ) {
     let parameters = *world.get_resource_or_insert_with(FlushParameters::default);
     let mut roster = OperationRoster::new();
     collect_from_channels(&parameters, new_service_query, world, &mut roster);
 
+    #[cfg(feature = "trace")]
+    {
+        let debug = debug.get(world);
+        if let Some(debug) = debug
+            && debug.is_changed()
+        {
+            world.get_resource_or_init::<DebugRoster>();
+            world.resource_scope::<DebugRoster, _>(|world, mut debug_roster| {
+                debug_roster.release_unpaused(world, &mut roster);
+            });
+        }
+
+        // Queue any operations that needed to be deferred
+        let mut deferred = world.get_resource_or_insert_with(DeferredRoster::default);
+        roster.append(&mut deferred);
+    }
+
     let mut loop_count = 0;
     while !roster.is_empty() {
         for e in roster.deferred_despawn.drain(..) {

src/flush.rs

@@ -35,7 +35,9 @@ use crate::{
 };
 
 #[cfg(feature = "trace")]
-use crate::{MessageSent, Trace, TraceToggle, TracedEvent, UniversalTraceToggle};
+use crate::{
+    Debug, DebugRoster, MessageSent, Trace, TraceToggle, TracedEvent, UniversalTraceToggle,
+};
 
 pub type Seq = u32;
 
@@ -486,9 +488,66 @@ impl ManageInput for World {
         &mut self,
         source: Entity,
     ) -> Result<Option<Input<T>>, OperationError> {
-        let mut storage = self.get_mut::<InputStorage<T>>(source).or_broken()?;
-        let input = storage.reverse_queue.pop();
-        Ok(input)
+        #[cfg(not(feature = "trace"))]
+        {
+            let mut storage = self.get_mut::<InputStorage<T>>(source).or_broken()?;
+            let input = storage.reverse_queue.pop();
+            return Ok(input);
+        }
+
+        #[cfg(feature = "trace")]
+        {
+            self.get_resource_or_init::<Debug>();
+            self.resource_scope::<Debug, _>(|world, mut debug| {
+                if !debug.is_active() {
+                    // Revert to the usual implementation of popping the next
+                    let mut storage = world.get_mut::<InputStorage<T>>(source).or_broken()?;
+                    let input = storage.reverse_queue.pop();
+                    return Ok(input);
+                } else {
+                    world.get_resource_or_init::<DebugRoster>();
+                    world.resource_scope::<DebugRoster, _>(|world, mut debug_roster| {
+                        let storage = world.get::<InputStorage<T>>(source).or_broken()?;
+                        let rev_next = storage.reverse_queue.iter().rev().position(|input| {
+                            let session = input.session;
+                            let seq = input.seq;
+
+                            // Evaluate whether we have hit a breakpoint. This will
+                            // pause the current session if we have.
+                            debug.evaluate_break(session, source, world);
+
+                            let mut is_paused = debug.is_paused(session, world);
+                            if is_paused {
+                                // We need to track this request inside the debug roster
+                                if debug_roster.is_allowed(RequestId {
+                                    session,
+                                    source,
+                                    seq,
+                                }) {
+                                    // If this input has been given permission to
+                                    // to be taken, change is_paused to false so
+                                    // it will be passed along.
+                                    is_paused = false;
+                                }
+                            }
+
+                            !is_paused
+                        });
+
+                        debug.notify_session_changes(world);
+
+                        if let Some(rev_next) = rev_next {
+                            let mut storage =
+                                world.get_mut::<InputStorage<T>>(source).or_broken()?;
+                            let next = storage.reverse_queue.len() - rev_next - 1;
+                            return Ok(Some(storage.reverse_queue.remove(next)));
+                        } else {
+                            return Ok(None);
+                        }
+                    })
+                }
+            })
+        }
     }
 
     fn cleanup_inputs<T: 'static + Send + Sync>(

src/input.rs

@@ -93,6 +93,11 @@ pub use chain::*;
 pub mod channel;
 pub use channel::*;
 
+#[cfg(feature = "trace")]
+pub mod debug;
+#[cfg(feature = "trace")]
+pub use debug::*;
+
 #[cfg(feature = "diagram")]
 pub mod diagram;
 #[cfg(feature = "diagram")]
@@ -515,6 +520,9 @@ pub mod prelude {
         },
     };
 
+    #[cfg(feature = "trace")]
+    pub use crate::{Debug, DebugStepExt, UniversalTraceToggle};
+
     pub use futures::FutureExt;
 }
 

src/lib.rs

@@ -220,7 +220,7 @@ impl OperationType {
     }
 }
 
-#[derive(Default)]
+#[derive(Default, Debug)]
 pub struct OperationRoster {
     /// Operation sources that should be triggered
     pub(crate) queue: VecDeque<Entity>,
@@ -364,7 +364,7 @@ impl OperationError {
     }
 }
 
-pub type OperationResult = Result<(), OperationError>;
+pub type OperationResult<T = ()> = Result<T, OperationError>;
 pub type ReachabilityResult = Result<bool, OperationError>;
 
 pub struct OperationSetup<'a> {

src/operation.rs

@@ -136,6 +136,32 @@ mod tests {
 
         let mut context = TestingContext::minimal_plugins();
 
+        #[cfg(feature = "trace")]
+        use {
+            crate::{TracedEvent, TracedEventKind},
+            bevy_ecs::prelude::{Resource, Trigger},
+        };
+
+        #[cfg(feature = "trace")]
+        #[derive(Debug, Resource, Default)]
+        struct TraceLog {
+            events: Vec<TracedEventKind>,
+        }
+
+        #[cfg(feature = "trace")]
+        {
+            context
+                .app
+                .world_mut()
+                .get_resource_or_insert_with(UniversalTraceToggle::on);
+            context.app.world_mut().add_observer(
+                |trace: Trigger<TracedEvent>, world: &mut World| {
+                    let mut log = world.get_resource_or_init::<TraceLog>();
+                    log.events.push(trace.event().event.clone());
+                },
+            );
+        }
+
         let workflow = context.spawn_io_workflow(|scope, builder| {
             let node = builder.create_map(|input: Async<f64, StreamOf<f64>>| async move {
                 input.streams.send(input.request);
@@ -158,7 +184,20 @@ mod tests {
                 .connect(scope.terminate);
         });
 
-        let r = context.resolve_request(2.0, workflow);
+        let r = match context.try_resolve_request(2.0, workflow, Duration::from_secs(2)) {
+            Ok(ok) => ok,
+            Err(err) => {
+                #[cfg(feature = "trace")]
+                {
+                    println!(
+                        "Activity trace of test failure:\n{:#?}",
+                        context.app.world().resource::<TraceLog>(),
+                    );
+                }
+                assert!(false, "failed to resolve:\n{err}");
+                return;
+            }
+        };
         assert_eq!(r, 4.0);
     }
 

src/operation/join.rs

@@ -15,7 +15,7 @@
  *
 */
 
-use bevy_ecs::prelude::{Component, Entity, World};
+use bevy_ecs::prelude::{ChildOf, Component, Entity, World};
 
 use crate::{Cancellation, ScopedSessionBundle, Seq};
 
@@ -69,7 +69,14 @@ pub trait ManageSession {
         seq: Seq,
     ) -> Entity;
 
+    /// Despawn a session
     fn despawn_session(&mut self, entity: Entity);
+
+    /// Returns true if `descendent_session` is a descendent of `parent_session`.
+    /// Note that this also returns true if `parent_session == descendent_session`,
+    /// so this is meant to be used when you're trying to figure out of a session
+    /// exists anywhere inside another session, inclusively.
+    fn is_descendent_session(&self, parent_session: Entity, descendent_session: Entity) -> bool;
 }
 
 impl ManageSession for World {
@@ -128,6 +135,26 @@ impl ManageSession for World {
 
         self.despawn(session);
     }
+
+    fn is_descendent_session(
+        &self,
+        parent_session: Entity,
+        mut descendent_session: Entity,
+    ) -> bool {
+        if parent_session == descendent_session {
+            return true;
+        }
+
+        while let Some(parent) = self.get::<ChildOf>(descendent_session).map(|c| c.parent()) {
+            if parent == parent_session {
+                return true;
+            }
+
+            descendent_session = parent;
+        }
+
+        false
+    }
 }
 
 #[cfg(feature = "trace")]

src/session.rs

@@ -486,6 +486,36 @@ impl SessionEvent {
         world.trigger(TracedEvent::now(event));
     }
 
+    pub(crate) fn paused_by_user(session: Entity, world: &mut World) {
+        let session_stack = get_session_stack_from_world(session, world);
+        let event = SessionEvent {
+            session_stack,
+            change: SessionChange::Paused(PauseCause::UserRequest),
+        };
+
+        world.trigger(TracedEvent::now(event));
+    }
+
+    pub(crate) fn paused_by_breakpoint(session: Entity, breakpoint: Entity, world: &mut World) {
+        let session_stack = get_session_stack_from_world(session, world);
+        let event = SessionEvent {
+            session_stack,
+            change: SessionChange::Paused(PauseCause::Breakpoint(breakpoint)),
+        };
+
+        world.trigger(TracedEvent::now(event));
+    }
+
+    pub(crate) fn unpaused(session: Entity, world: &mut World) {
+        let session_stack = get_session_stack_from_world(session, world);
+        let event = SessionEvent {
+            session_stack,
+            change: SessionChange::Unpaused,
+        };
+
+        world.trigger(TracedEvent::now(event));
+    }
+
     pub(crate) fn cleanup(session: Entity, world: &mut World) {
         let session_stack = get_session_stack_from_world(session, world);
         let event = SessionEvent {
@@ -513,6 +543,14 @@ pub enum SessionChange {
     },
     BeginCleanup,
     Despawned,
+    Paused(PauseCause),
+    Unpaused,
+}
+
+#[derive(Debug, Clone)]
+pub enum PauseCause {
+    UserRequest,
+    Breakpoint(Entity),
 }
 
 #[derive(Debug, Clone)]