Refactor pause state variables

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, paused_by: Optional[str] = None) -> 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, paused_by: Optional[str] = None) -> 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
         self.task_details.paused_by = paused_by if paused_by else ""
 
@@ -379,14 +386,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)
             self.task_details.paused_by = ""
         except TaskStartError as e:
             # start_task_async() already set status to ERROR, but goal_done is only ever set via the done-callback
@@ -401,14 +408,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,
@@ -465,7 +472,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()

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,25 +199,25 @@ 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])
         )
 
         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)