From 69444bc5511e8eca16e0949c6fcf8f600e48e8b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joni=20P=C3=B6ll=C3=A4nen?= Date: Fri, 11 Sep 2026 16:21:12 +0300 Subject: [PATCH] Refactor pause state variables --- task_manager/task_manager/task_client.py | 63 +++++++++++++----------- task_manager/test/test_task_clients.py | 30 ++++++----- 2 files changed, 49 insertions(+), 44 deletions(-) diff --git a/task_manager/task_manager/task_client.py b/task_manager/task_manager/task_client.py index 43e282e..f76a75c 100644 --- a/task_manager/task_manager/task_client.py +++ b/task_manager/task_manager/task_client.py @@ -15,6 +15,7 @@ # ------------------------------------------------------------------ from abc import ABC, abstractmethod +from dataclasses import dataclass, field from multiprocessing import Event from typing import Any, Callable, Dict, List, Optional @@ -90,8 +91,21 @@ def resume_task(self) -> None: """Resume a previously paused task synchronously.""" -# TODO: Bundle pausing related members to resolve too-many-instance-attributes -class ActionTaskClient(TaskClient): # pylint: disable=too-many-instance-attributes +@dataclass +class _PauseState: + """Bookkeeping for ActionTaskClient's pause/resume mechanics.""" + + pausing: bool = False + paused: bool = False + pause_done: Any = field(default_factory=Event) + last_goal_message: Optional[Any] = None + # Result of each paused segment (the Result the server returned when its goal was cancelled by + # pause_task()), in chronological order. Merged into the final result via + # task_specs.result_concat_fields once the task truly finishes. + paused_results: List[Any] = field(default_factory=list) + + +class ActionTaskClient(TaskClient): """Task client that keeps track of a single Action task.""" DONE_STATES = [GoalStatus.STATUS_SUCCEEDED, GoalStatus.STATUS_ABORTED, GoalStatus.STATUS_CANCELED] @@ -113,14 +127,7 @@ def __init__( self._task_details = task_details self._task_specs = task_specs self._goal_done = Event() - self._pausing = False - self._paused = False - self._pause_done = Event() - self._last_goal_message: Optional[Any] = None - # Result of each paused segment (the Result the server returned when its goal was cancelled by - # pause_task()), in chronological order. Merged into the final result via - # task_specs.result_concat_fields once the task truly finishes. - self._paused_results: List[Any] = [] + self._pause_state = _PauseState() self._task_done_callbacks: List[Callable[[TaskSpecs, TaskDetails], None]] = [] @@ -168,7 +175,7 @@ def start_task_async(self, goal_message: Any) -> None: :param goal_message: ROS action goal message :raises TaskStartError: If task cannot be started """ - self._last_goal_message = goal_message + self._pause_state.last_goal_message = goal_message if not self._client.wait_for_server(timeout_sec=self.server_wait_timeout): self.task_details.status = TaskStatus.ERROR @@ -263,9 +270,9 @@ def _finish_if_paused(self) -> bool: cancel's done-callback was suppressed by pause_task() at the time (see `_goal_done_cb`) - so nothing will ever fire it on its own; this is the only place that ever will. Returns False (no-op) otherwise. """ - if not self._paused: + if not self._pause_state.paused: return False - self._paused = False + self._pause_state.paused = False self.task_details.status = TaskStatus.CANCELED self.task_details.result = self._merged_result() self._notify_done_callbacks() @@ -340,24 +347,24 @@ def pause_task(self) -> None: :raises PauseTaskFailedError: If the task is already paused/finished, or the cancel fails/times out. """ - if self._paused: + if self._pause_state.paused: raise PauseTaskFailedError("Task is already paused.") if self.goal_done: raise PauseTaskFailedError("Cannot pause a task that has already finished.") if not self._goal_handle: raise PauseTaskFailedError("Couldn't pause the task, goal handle does not exist!") - self._pausing = True - self._pause_done = Event() + self._pause_state.pausing = True + self._pause_state.pause_done = Event() try: self.request_canceling() except CancelTaskFailedError as e: - self._pausing = False + self._pause_state.pausing = False raise PauseTaskFailedError(f"Failed to pause the task: {e}") from e # Wait for _goal_done_cb to set the pause_done event, which indicates that the cancel has been done. - if not self._pause_done.wait(timeout=self._task_specs.cancel_timeout): - self._pausing = False + if not self._pause_state.pause_done.wait(timeout=self._task_specs.cancel_timeout): + self._pause_state.pausing = False if self.goal_done: # The goal finished on its own (succeeded/aborted) before our cancel could take effect - # _goal_done_cb() has already fired the real done-callbacks with the real result. Nothing to undo, @@ -367,8 +374,8 @@ def pause_task(self) -> None: f"Task didn't pause within {self._task_specs.cancel_timeout} second timeout after it was cancelled." ) - self._pausing = False - self._paused = True + self._pause_state.pausing = False + self._pause_state.paused = True self.task_details.status = TaskStatus.PAUSED def resume_task(self) -> None: @@ -378,14 +385,14 @@ def resume_task(self) -> None: :raises ResumeTaskFailedError: If (re)starting the goal fails. """ - if not self._paused: + if not self._pause_state.paused: return - self._paused = False + self._pause_state.paused = False self._goal_handle = None self._result_future = None try: - self.start_task_async(self._last_goal_message) + self.start_task_async(self._pause_state.last_goal_message) except TaskStartError as e: # start_task_async() already set status to ERROR, but goal_done is only ever set via the done-callback # chain, which nothing else will trigger for a start failure - without this, the task would stay @@ -399,14 +406,14 @@ def _goal_done_cb(self, future: Future) -> None: :param future: Future object giving the result of the action call. """ - if self._pausing and self._canceled_result(future): + if self._pause_state.pausing and self._canceled_result(future): # This is the cancel triggered from pause_task(). The task is not actually # finished, so we skip the normal "task done" side effects entirely. The partial result the # server attached to the cancelled goal is kept, to be concatenated into the final result # (see _merged_result). if not future.cancelled(): - self._paused_results.append(future.result().result) - self._pause_done.set() + self._pause_state.paused_results.append(future.result().result) + self._pause_state.pause_done.set() return # Reached even while _pausing is True if the goal genuinely finished (succeeded/aborted) on its own, @@ -463,7 +470,7 @@ def _merged_result(self, final_result: Optional[Any] = None) -> Any: Best-effort: a listed field that doesn't exist or isn't concatenatable is logged and skipped, never fatal. """ - partials = self._paused_results + partials = self._pause_state.paused_results if final_result is None: if not partials: return self.task_specs.msg_interface.Result() diff --git a/task_manager/test/test_task_clients.py b/task_manager/test/test_task_clients.py index 4773aa3..46d44ea 100644 --- a/task_manager/test/test_task_clients.py +++ b/task_manager/test/test_task_clients.py @@ -127,9 +127,7 @@ def test_cancel_task_short_circuits_when_paused(self): """Cancelling an already-paused task must finish it directly, without touching the (gone) goal handle.""" task_client = get_action_task_client("task_1") task_client.register_done_callback(self._done_cb) - # `_paused` (not the public status) is the source of truth for "this client's own goal was really - # cancelled by pause_task()" - see _finish_if_paused()'s docstring for why. - task_client._paused = True + task_client._pause_state.paused = True task_client.task_details.status = TaskStatus.PAUSED task_client.cancel_task() @@ -145,7 +143,7 @@ def test_request_canceling_short_circuits_when_paused(self): """ task_client = get_action_task_client("task_1") task_client.register_done_callback(self._done_cb) - task_client._paused = True + task_client._pause_state.paused = True task_client.task_details.status = TaskStatus.PAUSED task_client.request_canceling() @@ -163,7 +161,7 @@ def test_cancel_task_does_not_short_circuit_a_composite_forced_into_paused_statu """ task_client = get_action_task_client("task_1") task_client.register_done_callback(self._done_cb) - task_client.task_details.status = TaskStatus.PAUSED # forced externally, _paused was never set + task_client.task_details.status = TaskStatus.PAUSED # forced externally, _pause_state.paused was never set task_client._goal_handle = Mock(status=GoalStatus.STATUS_EXECUTING) task_client._goal_done.set() # pretend the cancel completed instantly - only the dispatch is under test @@ -201,7 +199,7 @@ def test_pause_captures_partial_result(self): """The Result the server attaches to the pause-cancel must be stored, without any "task done" side effects.""" task_client = get_action_task_client("task_1") task_client.register_done_callback(self._done_cb) - task_client._pausing = True + task_client._pause_state.pausing = True goal_future = Future(executor=Mock()) goal_future._result = Fibonacci.Impl.GetResultService.Response( status=GoalStatus.STATUS_CANCELED, result=Fibonacci.Result(sequence=[0, 1, 1]) @@ -209,17 +207,17 @@ def test_pause_captures_partial_result(self): task_client._goal_done_cb(goal_future) - self.assertTrue(task_client._pause_done.is_set()) + self.assertTrue(task_client._pause_state.pause_done.is_set()) self.assertFalse(self.cb_called) self.assertFalse(task_client.goal_done) - self.assertEqual(len(task_client._paused_results), 1) - self.assertEqual(extract_values(task_client._paused_results[0]), {"sequence": [0, 1, 1]}) + self.assertEqual(len(task_client._pause_state.paused_results), 1) + self.assertEqual(extract_values(task_client._pause_state.paused_results[0]), {"sequence": [0, 1, 1]}) def test_final_result_concatenates_paused_segments(self): """Fields listed in result_concat_fields are concatenated across paused segments and the final segment, in chronological order.""" task_client = get_concat_action_task_client(result_concat_fields=["sequence"]) - task_client._paused_results = [Fibonacci.Result(sequence=[0, 1]), Fibonacci.Result(sequence=[1, 2])] + task_client._pause_state.paused_results = [Fibonacci.Result(sequence=[0, 1]), Fibonacci.Result(sequence=[1, 2])] goal_future = Future(executor=Mock()) goal_future._result = Fibonacci.Impl.GetResultService.Response( status=GoalStatus.STATUS_SUCCEEDED, result=Fibonacci.Result(sequence=[3, 5]) @@ -234,7 +232,7 @@ def test_final_result_concatenates_paused_segments(self): def test_final_result_without_concat_fields_keeps_final_segment_only(self): """With no result_concat_fields configured, paused segments' results are discarded as before.""" task_client = get_concat_action_task_client(result_concat_fields=[]) - task_client._paused_results = [Fibonacci.Result(sequence=[0, 1])] + task_client._pause_state.paused_results = [Fibonacci.Result(sequence=[0, 1])] goal_future = Future(executor=Mock()) goal_future._result = Fibonacci.Impl.GetResultService.Response( status=GoalStatus.STATUS_SUCCEEDED, result=Fibonacci.Result(sequence=[3, 5]) @@ -247,7 +245,7 @@ def test_final_result_without_concat_fields_keeps_final_segment_only(self): def test_final_result_skips_bad_concat_field(self): """A configured field that can't be concatenated is skipped without breaking the rest of the result.""" task_client = get_concat_action_task_client(result_concat_fields=["no_such_field", "sequence"]) - task_client._paused_results = [Fibonacci.Result(sequence=[0, 1])] + task_client._pause_state.paused_results = [Fibonacci.Result(sequence=[0, 1])] goal_future = Future(executor=Mock()) goal_future._result = Fibonacci.Impl.GetResultService.Response( status=GoalStatus.STATUS_SUCCEEDED, result=Fibonacci.Result(sequence=[3, 5]) @@ -261,9 +259,9 @@ def test_final_result_skips_bad_concat_field(self): def test_cancel_while_paused_reports_merged_paused_segments(self): """Cancelling a paused task must report the merged partial results instead of an empty Result.""" task_client = get_concat_action_task_client(result_concat_fields=["sequence"]) - task_client._paused = True + task_client._pause_state.paused = True task_client.task_details.status = TaskStatus.PAUSED - task_client._paused_results = [Fibonacci.Result(sequence=[0, 1]), Fibonacci.Result(sequence=[1, 2])] + task_client._pause_state.paused_results = [Fibonacci.Result(sequence=[0, 1]), Fibonacci.Result(sequence=[1, 2])] task_client.cancel_task() @@ -273,8 +271,8 @@ def test_cancel_while_paused_reports_merged_paused_segments(self): def test_resume_failure_reports_merged_paused_segments(self): """A paused task whose restart fails must still report the merged partial results.""" task_client = get_concat_action_task_client(result_concat_fields=["sequence"]) - task_client._paused = True - task_client._paused_results = [Fibonacci.Result(sequence=[0, 1])] + task_client._pause_state.paused = True + task_client._pause_state.paused_results = [Fibonacci.Result(sequence=[0, 1])] with patch.object(ActionTaskClient, "start_task_async", side_effect=TaskStartError("server gone")): self.assertRaises(ResumeTaskFailedError, task_client.resume_task)