Properly backport #947 to fix callback shutdown deadlock (humble)

rclpy/rclpy/executors.py

@@ -82,15 +82,15 @@ def __exit__(self, t, v, tb):
             self._num_work_executing -= 1
             self._work_condition.notify_all()
 
-    def wait(self, timeout_sec=None):
+    def wait(self, timeout_sec: Optional[float] = None):
         """
         Wait until all work completes.
 
         :param timeout_sec: Seconds to wait. Block forever if None or negative. Don't wait if 0
         :type timeout_sec: float or None
         :rtype: bool True if all work completed
         """
-        if timeout_sec is not None and timeout_sec < 0:
+        if timeout_sec is not None and timeout_sec < 0.0:
             timeout_sec = None
         # Wait for all work to complete
         with self._work_condition:
@@ -240,17 +240,13 @@ def shutdown(self, timeout_sec: float = None) -> bool:
             timeot expires before all outstanding work is done.
         """
         with self._shutdown_lock:
-            if self._is_shutdown:
-                return True
-            self._is_shutdown = True
-            # Tell executor it's been shut down
-            try:
+            if not self._is_shutdown:
+                self._is_shutdown = True
+                # Tell executor it's been shut down
                 self._guard.trigger()
-            except InvalidHandle:
-                pass
-
-        if not self._work_tracker.wait(timeout_sec):
-            return False
+        if not self._is_shutdown:
+            if not self._work_tracker.wait(timeout_sec):
+                return False
 
         # Clean up stuff that won't be used anymore
         with self._nodes_lock:
@@ -469,27 +465,24 @@ async def handler(entity, gc, is_shutdown, work_tracker):
             if is_shutdown or not entity.callback_group.beginning_execution(entity):
                 # Didn't get the callback, or the executor has been ordered to stop
                 entity._executor_event = False
-                try:
-                    gc.trigger()
-                except InvalidHandle:
-                    pass
+                gc.trigger()
                 return
             with work_tracker:
                 arg = take_from_wait_list(entity)
 
                 # Signal that this has been 'taken' and can be added back to the wait list
                 entity._executor_event = False
-                try:
-                    gc.trigger()
-                except InvalidHandle:
-                    pass
+                gc.trigger()
 
                 try:
                     await call_coroutine(entity, arg)
                 finally:
                     entity.callback_group.ending_execution(entity)
                     # Signal that work has been done so the next callback in a mutually exclusive
                     # callback group can get executed
+
+                    # Catch expected error where calling executor.shutdown()
+                    # from callback causes the GuardCondition to be destroyed
                     try:
                         gc.trigger()
                     except InvalidHandle:
@@ -581,10 +574,7 @@ def _wait_for_ready_callbacks(
                 # retrigger a guard condition that was triggered but not handled
                 for gc in node_guards:
                     if gc._executor_triggered:
-                        try:
-                            gc.trigger()
-                        except InvalidHandle:
-                            pass
+                        gc.trigger()
                     guards.append(gc)
             if timeout_timer is not None:
                 timers.append(timeout_timer)

rclpy/test/test_executor.py

@@ -576,38 +576,24 @@ def timer2_callback():
         timer1.destroy()
         cli.destroy()
 
-    def test_shutdown_from_callback_no_deadlock(self):
-        test_context = rclpy.context.Context()
-        rclpy.init(context=test_context)
-
-        try:
-            test_node = rclpy.create_node('test_shutdown_node', context=test_context)
-            shutdown_called = [False]
-
-            def timer_callback():
-                shutdown_called[0] = True
-                rclpy.shutdown(context=test_context)
-
-            timer = test_node.create_timer(0.1, timer_callback)
-
-            executor = SingleThreadedExecutor(context=test_context)
-            executor.add_node(test_node)
-
-            start_time = time.monotonic()
-            while not shutdown_called[0] and time.monotonic() - start_time < 5.0:
-                executor.spin_once(timeout_sec=0.1)
-
-            self.assertTrue(shutdown_called[0], 'Timer callback was not executed')
+    def test_shutdown_executor_from_callback(self):
+        """https://github.com/ros2/rclpy/issues/944: allow for executor shutdown from callback."""
+        self.assertIsNotNone(self.node.handle)
+        timer_period = 0.1
+        executor = SingleThreadedExecutor(context=self.context)
+        shutdown_event = threading.Event()
 
-            test_node.destroy_timer(timer)
-            test_node.destroy_node()
+        def timer_callback():
+            nonlocal shutdown_event, executor
             executor.shutdown()
+            shutdown_event.set()
 
-        finally:
-            try:
-                rclpy.shutdown(context=test_context)
-            except Exception:
-                pass
+        tmr = self.node.create_timer(timer_period, timer_callback)
+        executor.add_node(self.node)
+        t = threading.Thread(target=executor.spin, daemon=True)
+        t.start()
+        self.assertTrue(shutdown_event.wait(120))
+        self.node.destroy_timer(tmr)
 
 
 if __name__ == '__main__':