Implement uninterruptibility for scopes

Signed-off-by: Michael X. Grey <grey@openrobotics.org>

Author: mxgrey

src/builder.rs

@@ -123,7 +123,8 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         Streams: StreamPack,
     {
         let scope_id = self.commands.spawn(()).id();
-        self.create_scope_impl(scope_id, settings, build)
+        let exit_scope = self.commands.spawn(UnusedTarget).id();
+        self.create_scope_impl(scope_id, exit_scope, settings, build)
     }
 
     /// It is possible for a scope to be cancelled before it terminates. Even a
@@ -152,32 +153,21 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         build: impl FnOnce(Scope<T, (), ()>, &mut Builder),
     ) {
         let cancelling_scope_id = self.commands.spawn(()).id();
-        let cancelling_operation = OperateScope::<T, (), ()>::new(
-            cancelling_scope_id, Some(self.finish_scope_cancel), settings, self.commands,
+        let _ = self.create_scope_impl::<T, (), ()>(
+            cancelling_scope_id,
+            self.finish_scope_cancel,
+            settings,
+            build,
         );
-        self.commands.add(AddOperation::new(cancelling_scope_id, cancelling_operation));
 
         let begin_cancel = self.commands.spawn(()).id();
         self.commands.add(AddOperation::new(
             begin_cancel,
             BeginCancel::<T>::new(self.scope, from_buffer.source, cancelling_scope_id),
         ));
-        let mut builder = Builder {
-            scope: cancelling_scope_id,
-            finish_scope_cancel: cancelling_operation.finish_cancel(),
-            commands: self.commands,
-        };
-
-        let scope = Scope {
-            input: Output::new(cancelling_scope_id, cancelling_operation.enter_scope()),
-            terminate: InputSlot::new(cancelling_scope_id, cancelling_operation.terminal()),
-            streams: (),
-        };
-
-        build(scope, &mut builder);
     }
 
-    /// Get the scope that this builder is building for
+    /// Get the scope that this builder is building for.
     pub fn scope(&self) -> Entity {
         self.scope
     }
@@ -191,6 +181,7 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
     pub(crate) fn create_scope_impl<Request, Response, Streams>(
         &mut self,
         scope_id: Entity,
+        exit_scope: Entity,
         settings: ScopeSettings,
         build: impl FnOnce(Scope<Request, Response, Streams>, &mut Builder),
     ) -> Node<Request, Response, Streams>
@@ -199,10 +190,12 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
         Response: 'static + Send + Sync,
         Streams: StreamPack,
     {
-        let exit_scope = self.commands.spawn(UnusedTarget).id();
         let operation = OperateScope::<Request, Response, Streams>::new(
             scope_id, Some(exit_scope), settings, self.commands,
         );
+        let enter_scope = operation.enter_scope();
+        let finish_scope_cancel = operation.finish_cancel();
+        let terminal = operation.terminal();
         self.commands.add(AddOperation::new(scope_id, operation));
 
         let (stream_in, stream_out) = Streams::spawn_scope_streams(
@@ -213,13 +206,13 @@ impl<'w, 's, 'a> Builder<'w, 's, 'a> {
 
         let mut builder = Builder {
             scope: scope_id,
-            finish_scope_cancel: operation.finish_cancel(),
+            finish_scope_cancel,
             commands: self.commands,
         };
 
         let scope = Scope {
-            input: Output::new(scope_id, operation.enter_scope()),
-            terminate: InputSlot::new(scope_id, operation.terminal()),
+            input: Output::new(scope_id, enter_scope),
+            terminate: InputSlot::new(scope_id, terminal),
             streams: stream_in,
         };
 

src/chain.rs

@@ -231,8 +231,9 @@ impl<'w, 's, 'a, 'b, T: 'static + Send + Sync> Chain<'w, 's, 'a, 'b, T> {
         Response: 'static + Send + Sync,
         Streams: StreamPack,
     {
+        let exit_scope = self.builder.commands.spawn(UnusedTarget).id();
         self.builder.create_scope_impl::<T, Response, Streams>(
-            self.target, settings, build,
+            self.target, exit_scope, settings, build,
         ).output.chain(self.builder)
     }
 
@@ -248,8 +249,9 @@ impl<'w, 's, 'a, 'b, T: 'static + Send + Sync> Chain<'w, 's, 'a, 'b, T> {
         Response: 'static + Send + Sync,
         Streams: StreamPack,
     {
+        let exit_scope = self.builder.commands.spawn(UnusedTarget).id();
         self.builder.create_scope_impl::<T, Response, Streams>(
-            self.target, settings, build,
+            self.target, exit_scope, settings, build,
         )
     }
 

src/operation/scope.rs

@@ -61,6 +61,8 @@ pub(crate) struct OperateScope<Request, Response, Streams> {
     exit_scope: Option<Entity>,
     /// Cancellation finishes at this node
     finish_cancel: Entity,
+    /// Settings for the scope
+    settings: ScopeSettings,
     _ignore: std::marker::PhantomData<(Request, Response, Streams)>,
 }
 
@@ -71,13 +73,12 @@ impl<Request, Response, Streams> Clone for OperateScope<Request, Response, Strea
             terminal: self.terminal,
             exit_scope: self.exit_scope,
             finish_cancel: self.finish_cancel,
+            settings: self.settings.clone(),
             _ignore: Default::default(),
         }
     }
 }
 
-impl<Request, Response, Streams> Copy for OperateScope<Request, Response, Streams> {}
-
 impl<Request, Response, Streams> OperateScope<Request, Response, Streams> {
     pub(crate) fn terminal(&self) -> Entity {
         self.terminal
@@ -107,22 +108,35 @@ impl ScopedSession {
     }
 }
 
+#[derive(Component)]
+struct ScopeSettingsStorage(ScopeSettings);
+
 #[derive(Component)]
 pub(crate) enum ScopedSessionStatus {
     Ongoing,
     Finished,
+    /// The scope was asked to cleanup from an external source, but it has an
+    /// uninterruptible setting. We are waiting for a termination or internal
+    /// cancel to trigger before doing a cleanup.
+    DeferredCleanup,
+    /// The scope has already begun the cleanup process
     Cleanup,
     Cancelled(Cancellation),
 }
 
 impl ScopedSessionStatus {
-    fn to_cleanup(&mut self) -> bool {
+    fn to_cleanup(&mut self, uninterruptible: bool) -> bool {
         if matches!(self, Self::Cleanup) {
             return false;
         }
 
-        *self = Self::Cleanup;
-        true
+        if uninterruptible {
+            *self = Self::DeferredCleanup;
+            return false;
+        } else {
+            *self = Self::Cleanup;
+            return true;
+        }
     }
 
     pub(crate) fn to_finished(&mut self) -> bool {
@@ -132,11 +146,19 @@ impl ScopedSessionStatus {
             *self = Self::Finished;
             return true;
         }
+
+        if matches!(self, Self::DeferredCleanup) {
+            // We've been waiting for the scope to finish before beginning
+            // cleanup because the scope is uninterruptible.
+            *self = Self::Cleanup;
+            return true;
+        }
+
         false
     }
 
     fn to_cancelled(&mut self, cancellation: Cancellation) -> bool {
-        if matches!(self, Self::Ongoing) {
+        if matches!(self, Self::Ongoing | Self::DeferredCleanup) {
             *self = Self::Cancelled(cancellation);
             return true;
         }
@@ -169,7 +191,6 @@ where
         source_mut.insert((
             InputBundle::<Request>::new(),
             ScopeEntryStorage(self.enter_scope),
-            FinishedStagingStorage(self.terminal),
             ScopeContents::new(),
             ScopedSessionStorage::default(),
             TerminalStorage(self.terminal),
@@ -178,6 +199,7 @@ where
             FinalizeScopeCleanup(Self::finalize_scope_cleanup),
             BeginCancelStorage::default(),
             FinishCancelStorage(self.finish_cancel),
+            ScopeSettingsStorage(self.settings),
         ));
 
         if let Some(exit_scope) = self.exit_scope {
@@ -218,14 +240,17 @@ where
         OperationCleanup { source, session, world, roster }: OperationCleanup
     ) -> OperationResult {
         let mut source_mut = world.get_entity_mut(source).or_broken()?;
+        let uninterruptible = source_mut.get::<ScopeSettingsStorage>().or_broken()?.0
+            .is_uninterruptible();
+
         let pairs: SmallVec<[_; 16]> = source_mut
             .get_mut::<ScopedSessionStorage>()
             .or_broken()?
             .0
             .iter_mut()
             .filter(|pair| pair.parent_session == session)
             .filter_map(|p| {
-                if p.status.to_cleanup() {
+                if p.status.to_cleanup(uninterruptible) {
                     Some(p.scoped_session)
                 } else {
                     None
@@ -241,15 +266,15 @@ where
 
         for scoped_session in pairs {
             let source_ref = world.get_entity(source).or_broken()?;
-            let staging_node = source_ref.get::<FinishedStagingStorage>().or_broken()?.0;
+            let terminal = source_ref.get::<TerminalStorage>().or_broken()?.0;
             let nodes = source_ref.get::<ScopeContents>().or_broken()?.nodes().clone();
             for node in nodes {
                 OperationCleanup { source: node, session: scoped_session, world, roster }.clean();
             }
 
             // OperateScope::cleanup gets called when the entire scope is being cancelled
             // so we need to clear out the staging node as well.
-            OperationCleanup { source: staging_node, session: scoped_session, world, roster }.clean();
+            OperationCleanup { source: terminal, session: scoped_session, world, roster }.clean();
         }
 
         Ok(())
@@ -261,7 +286,6 @@ where
         }
 
         let source_ref = reachability.world.get_entity(reachability.source).or_broken()?;
-        let staging = source_ref.get::<FinishedStagingStorage>().or_broken()?.0;
 
         if let Some(pair) = source_ref
             .get::<ScopedSessionStorage>().or_broken()?
@@ -272,13 +296,9 @@ where
                 pair.scoped_session,
                 reachability.source,
                 reachability.world,
-                &mut visited
+                &mut visited,
             );
 
-            if scoped_reachability.check_upstream(staging)? {
-                return Ok(true);
-            }
-
             let terminal = source_ref.get::<TerminalStorage>().or_broken()?.0;
             if scoped_reachability.check_upstream(terminal)? {
                 return Ok(true);
@@ -343,6 +363,7 @@ where
             terminal,
             exit_scope,
             finish_cancel,
+            settings,
             _ignore: Default::default(),
         };
 
@@ -478,8 +499,14 @@ where
                 // so we'll return a broken error here.
                 None.or_broken()?;
             }
+            ScopedSessionStatus::DeferredCleanup => {
+                // We shouldn't be in this function if the session is in a
+                // deferred cleanup state. We should be waiting for a finish
+                // or a cancellation to occur.
+                None.or_broken()?;
+            }
             ScopedSessionStatus::Finished => {
-                let staging = source_mut.get::<FinishedStagingStorage>().or_broken()?.0;
+                let terminal = source_mut.get::<TerminalStorage>().or_broken()?.0;
                 let (target, blocker) = source_mut.get_mut::<ExitTargetStorage>()
                     .and_then(|mut storage| storage.map.remove(&scoped_session))
                     .map(|exit| (exit.target, exit.blocker))
@@ -491,7 +518,7 @@ where
                     .or_broken()?;
 
                 let response = clean.world
-                    .get_mut::<Staging<Response>>(staging).or_broken()?.0
+                    .get_mut::<Staging<Response>>(terminal).or_broken()?.0
                     .remove(&clean.session).or_broken()?;
                 clean.world.get_entity_mut(target).or_broken()?.give_input(
                     pair.parent_session, response, clean.roster,
@@ -564,15 +591,6 @@ pub struct FinalizeScopeCleanup(pub(crate) fn(OperationCleanup) -> OperationResu
 #[derive(Component)]
 struct ScopeEntryStorage(Entity);
 
-#[derive(Component)]
-pub struct FinishedStagingStorage(Entity);
-
-impl FinishedStagingStorage {
-    pub fn get(&self) -> Entity {
-        self.0
-    }
-}
-
 pub(crate) struct Terminate<T> {
     _ignore: std::marker::PhantomData<T>,
 }
@@ -600,6 +618,7 @@ where
     fn setup(self, OperationSetup { source, world }: OperationSetup) -> OperationResult {
         world.entity_mut(source).insert((
             InputBundle::<T>::new(),
+            SingleInputStorage::empty(),
             Staging::<T>::new(),
         ));
         Ok(())
@@ -651,13 +670,17 @@ where
         Ok(())
     }
 
-    fn is_reachable(reachability: OperationReachability) -> ReachabilityResult {
+    fn is_reachable(mut reachability: OperationReachability) -> ReachabilityResult {
         if reachability.has_input::<T>()? {
             return Ok(true);
         }
 
         let staging = reachability.world.get::<Staging<T>>(reachability.source).or_broken()?;
-        Ok(staging.0.contains_key(&reachability.session))
+        if staging.0.contains_key(&reachability.session) {
+            return Ok(true);
+        }
+
+        SingleInputStorage::is_reachable(&mut reachability)
     }
 }
 

src/workflow.rs

@@ -153,7 +153,7 @@ pub enum DeliverySettings {
 }
 
 /// Settings which determine how the top-level scope of the workflow behaves.
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct ScopeSettings {
     /// Should we prevent the scope from being interrupted (e.g. cancelled)?
     /// False by default, meaning by default scopes can be cancelled or
@@ -198,18 +198,21 @@ impl<'w, 's> SpawnWorkflow for Commands<'w, 's> {
         let scope = OperateScope::<Request, Response, Streams>::new(
             scope_id, None, settings.scope, self,
         );
+        let enter_scope = scope.enter_scope();
+        let finish_scope_cancel = scope.finish_cancel();
+        let terminal = scope.terminal();
         self.add(AddOperation::new(scope_id, scope));
         let mut builder = Builder {
             scope: scope_id,
-            finish_scope_cancel: scope.finish_cancel(),
+            finish_scope_cancel,
             commands: self
         };
 
         let streams = Streams::spawn_workflow_streams(&mut builder);
 
         let scope = Scope {
-            input: Output::new(scope_id, scope.enter_scope()),
-            terminate: InputSlot::new(scope_id, scope.terminal()),
+            input: Output::new(scope_id, enter_scope),
+            terminate: InputSlot::new(scope_id, terminal),
             streams,
         };