Throw CancelledError into cancelled task

Signed-off-by: Nadav Elkabets <[email protected]>

Author: nadavelkabets

rclpy/rclpy/experimental/asyncio_executor.py

@@ -132,7 +132,6 @@ def __init__(self, timer: Timer, loop: asyncio.AbstractEventLoop, schedule_cb) -
         self._schedule_cb = schedule_cb
         self._loop = loop
         self._build_waiter()
-        self._timer.set_on_reset_callback(self.on_reset)
 
     def _build_waiter(self) -> None:
         self._waiter = _WaitHandler(
@@ -353,7 +352,7 @@ async def spin_once_async(
                 try:
                     await ready_task_getter
                 except asyncio.CancelledError:
-                    return
+                    pass
                 return
 
             if self._shutdown_fut in done:

rclpy/rclpy/task.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from asyncio import CancelledError
 from enum import Enum
 import inspect
 import sys
@@ -278,7 +279,7 @@ def __init__(self,
         self._task_lock = threading.Lock()
         self._fut_waiter: Optional[Future] = None
 
-    def __call__(self) -> None:
+    def __call__(self, *, should_cancel: bool = False) -> None:
         """
         Run or resume a task.
 
@@ -302,7 +303,10 @@ def __call__(self) -> None:
                 # Execute a coroutine
                 handler = self._handler
                 try:
-                    future = handler.send(None)
+                    if should_cancel:
+                        future = handler.throw(CancelledError())
+                    else:
+                        future = handler.send(None)
                     executor = self._executor()
                     if executor and hasattr(executor, '_resume_task'):
                         if future:
@@ -314,14 +318,19 @@ def __call__(self) -> None:
                     # The coroutine finished; store the result
                     self.set_result(e.value)
                     self._complete_task()
+                except CancelledError:
+                    super().cancel()
                 except Exception as e:
                     self.set_exception(e)
                     self._complete_task()
             else:
                 # Execute a normal function
                 try:
                     assert self._handler is not None and callable(self._handler)
-                    self.set_result(self._handler(*self._args, **self._kwargs))
+                    if should_cancel:
+                        super().cancel()
+                    else:
+                        self.set_result(self._handler(*self._args, **self._kwargs))
                 except Exception as e:
                     self.set_exception(e)
                 self._complete_task()
@@ -333,7 +342,7 @@ def __call__(self) -> None:
     def __wake(self, fut: Future):
         self._fut_waiter = None
         if fut.cancelled():
-            self.cancel()
+            self(should_cancel=True)
         elif fut.exception() is not None:
             self.set_exception(fut.exception())
         else:
@@ -358,7 +367,4 @@ def cancel(self) -> None:
             self._fut_waiter.cancel()
             return
 
-        if self._pending() and inspect.iscoroutine(self._handler):
-            self._handler.close()
-
-        super().cancel()
+        self(should_cancel=True)

rclpy/test/test_task.py

@@ -205,6 +205,23 @@ def test_executing(self) -> None:
         t = Task(lambda: None)
         self.assertFalse(t.executing())
 
+    def test_task_throws_cancelled_error_inside_coroutine_when_cancelled(self) -> None:
+        did_throw = False
+
+        async def test_coro():
+            try:
+                await Future()
+            except asyncio.CancelledError:
+                nonlocal did_throw
+                did_throw = True
+                raise
+
+        t = Task(test_coro)
+        t()
+        t.cancel()
+        self.assertTrue(t.cancelled())
+        self.assertTrue(did_throw)
+
 
 class TestFuture(unittest.TestCase):