Change timeout implementation and add unit test for timeout

Author: Juliaj

src/rai_core/rai/tools/__init__.py

@@ -12,4 +12,5 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from .timeout import timeout, timeout_method
+from .timeout import timeout as timeout
+from .timeout import timeout_method as timeout_method

src/rai_core/rai/tools/ros2/detection/tools.py

@@ -22,6 +22,7 @@
     PointCloudFilter,
     PointCloudFromSegmentation,
 )
+from rai.tools.timeout import TimeoutError, timeout
 
 
 class GetGrippingPointToolInput(BaseModel):
@@ -32,7 +33,6 @@ class GetGrippingPointToolInput(BaseModel):
 
 
 # TODO(maciejmajek): Configuration system configurable with namespacing
-# TODO(juliajia): Question for Maciej: for comments above on configuration system with namespacing, can you provide an use case for this?
 class GetGrippingPointTool(BaseROS2Tool):
     name: str = "get_gripping_point"
     description: str = "Get gripping points for specified object/objects. Returns 3D coordinates where a robot gripper can grasp the object."
@@ -68,10 +68,10 @@ def model_post_init(self, __context: Any) -> None:
 
     def _run(self, object_name: str) -> str:
         # this will be not work in agent scenario because signal need to be run in main thread, comment out for now
-        # @timeout(
-        #     self.timeout_sec,
-        #     f"Gripping point detection for object '{object_name}' exceeded {self.timeout_sec} seconds",
-        # )
+        @timeout(
+            self.timeout_sec,
+            f"Gripping point detection for object '{object_name}' exceeded {self.timeout_sec} seconds",
+        )
         def _run_with_timeout():
             pcl = self.point_cloud_from_segmentation.run(object_name)
             if len(pcl) == 0:
@@ -101,7 +101,7 @@ def _run_with_timeout():
 
         try:
             return _run_with_timeout()
-        except Exception as e:
-            if "timed out" in str(e).lower():
-                return f"Timeout: Gripping point detection for object '{object_name}' exceeded {self.timeout_sec} seconds"
+        except TimeoutError:
+            return f"Timeout: Gripping point detection for object '{object_name}' exceeded {self.timeout_sec} seconds"
+        except Exception:
             raise

src/rai_core/rai/tools/timeout.py

@@ -12,7 +12,28 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import signal
+"""
+Design considerations:
+
+Primary use case:
+- 3D object detection pipeline (image → point cloud → segmentation → gripping points)
+- Timeout long-running ROS2 service calls from agent tools
+
+RAI concurrency model:
+- `multiprocessing`: Process isolation (ROS2 launch)
+- `threading`: Agent execution and callbacks (LangChain agents in worker threads)
+- `asyncio`: Limited ROS2 coordination (LaunchManager)
+
+Timeout implementation:
+- Signal-based (SIGALRM): Only works in main thread, unsuitable for RAI's worker threads
+- ThreadPoolExecutor: Works in any thread, provides clean resource management
+
+Alternatives considered:
+- asyncio.wait_for(): Requires async context, conflicts with sync tool interface
+- threading.Timer: Potential resource leaks, less robust cleanup
+"""
+
+import concurrent.futures
 from functools import wraps
 from typing import Any, Callable, TypeVar
 
@@ -63,23 +84,16 @@ def timeout(seconds: float, timeout_message: str = None) -> Callable[[F], F]:
     def decorator(func: F) -> F:
         @wraps(func)
         def wrapper(*args, **kwargs):
-            def timeout_handler(signum, frame):
-                message = (
-                    timeout_message
-                    or f"Function '{func.__name__}' timed out after {seconds} seconds"
-                )
-                raise TimeoutError(message)
-
-            # Set up timeout
-            old_handler = signal.signal(signal.SIGALRM, timeout_handler)
-            signal.alarm(int(seconds))
-
-            try:
-                return func(*args, **kwargs)
-            finally:
-                # Clean up timeout
-                signal.alarm(0)
-                signal.signal(signal.SIGALRM, old_handler)
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(func, *args, **kwargs)
+                try:
+                    return future.result(timeout=seconds)
+                except concurrent.futures.TimeoutError:
+                    message = (
+                        timeout_message
+                        or f"Function '{func.__name__}' timed out after {seconds} seconds"
+                    )
+                    raise TimeoutError(message)
 
         return wrapper
 
@@ -116,23 +130,16 @@ def timeout_method(seconds: float, timeout_message: str = None) -> Callable[[F],
     def decorator(func: F) -> F:
         @wraps(func)
         def wrapper(self, *args, **kwargs):
-            def timeout_handler(signum, frame):
-                message = (
-                    timeout_message
-                    or f"Method '{func.__name__}' of {self.__class__.__name__} timed out after {seconds} seconds"
-                )
-                raise TimeoutError(message)
-
-            # Set up timeout
-            old_handler = signal.signal(signal.SIGALRM, timeout_handler)
-            signal.alarm(int(seconds))
-
-            try:
-                return func(self, *args, **kwargs)
-            finally:
-                # Clean up timeout
-                signal.alarm(0)
-                signal.signal(signal.SIGALRM, old_handler)
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(func, self, *args, **kwargs)
+                try:
+                    return future.result(timeout=seconds)
+                except concurrent.futures.TimeoutError:
+                    message = (
+                        timeout_message
+                        or f"Method '{func.__name__}' of {self.__class__.__name__} timed out after {seconds} seconds"
+                    )
+                    raise TimeoutError(message)
 
         return wrapper
 

tests/tools/ros2/test_detection_tools.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 import pytest
 
 try:
@@ -149,70 +151,42 @@ def test_point_cloud_filter():
 
 
 def test_get_gripping_point_tool_timeout():
-    """Test GetGrippingPointTool timeout behavior."""
-    # Mock the connector and components
+    # Complete mock setup
     mock_connector = Mock(spec=ROS2Connector)
-
-    # Create mock components that will simulate timeout
     mock_pcl_gen = Mock(spec=PointCloudFromSegmentation)
-    mock_pcl_gen.run.side_effect = lambda x: []  # Return empty to simulate no detection
-
     mock_filter = Mock(spec=PointCloudFilter)
-    mock_filter.run.return_value = []
-
     mock_estimator = Mock(spec=GrippingPointEstimator)
+
+    # Test 1: No timeout (fast execution)
+    mock_pcl_gen.run.return_value = []
+    mock_filter.run.return_value = []
     mock_estimator.run.return_value = []
 
-    # Create tool with short timeout
     tool = GetGrippingPointTool(
         connector=mock_connector,
-        point_cloud_from_segmentation=mock_pcl_gen,
-        point_cloud_filter=mock_filter,
+        target_frame="base",
+        source_frame="camera",
+        camera_topic="/image",
+        depth_topic="/depth",
+        camera_info_topic="/info",
         gripping_point_estimator=mock_estimator,
-        timeout_sec=0.1,
+        point_cloud_filter=mock_filter,
+        timeout_sec=5.0,
     )
+    tool.point_cloud_from_segmentation = mock_pcl_gen  # Connect the mock
 
-    # Test successful run with no gripping points found
-    result = tool._run("test_object")
-    assert "No gripping point found" in result
-    assert "test_object" in result
-
-    # Test with mock that simulates found gripping points
-    mock_estimator.run.return_value = [np.array([1.0, 2.0, 3.0], dtype=np.float32)]
-    result = tool._run("test_object")
-    assert "gripping point of the object test_object is" in result
-    assert "[1. 2. 3.]" in result
-
-    # Test with multiple gripping points
-    mock_estimator.run.return_value = [
-        np.array([1.0, 2.0, 3.0], dtype=np.float32),
-        np.array([4.0, 5.0, 6.0], dtype=np.float32),
-    ]
+    # Test fast execution - should complete without timeout
     result = tool._run("test_object")
-    assert "Multiple gripping points found" in result
-
-
-def test_get_gripping_point_tool_validation():
-    """Test GetGrippingPointTool input validation."""
-    mock_connector = Mock(spec=ROS2Connector)
-    mock_pcl_gen = Mock(spec=PointCloudFromSegmentation)
-    mock_filter = Mock(spec=PointCloudFilter)
-    mock_estimator = Mock(spec=GrippingPointEstimator)
-
-    # Test tool creation
-    tool = GetGrippingPointTool(
-        connector=mock_connector,
-        point_cloud_from_segmentation=mock_pcl_gen,
-        point_cloud_filter=mock_filter,
-        gripping_point_estimator=mock_estimator,
-    )
+    assert "No test_objects detected" in result
+    assert "timed out" not in result.lower()
 
-    # Verify tool properties
-    assert tool.name == "get_gripping_point"
-    assert "gripping points" in tool.description
-    assert tool.timeout_sec == 10.0  # default value
+    # Test 2: Actual timeout behavior
+    def slow_operation(obj_name):
+        time.sleep(2.0)  # Longer than timeout
+        return []
 
-    # Test args schema
-    from rai.tools.ros2.detection.tools import GetGrippingPointToolInput
+    mock_pcl_gen.run.side_effect = slow_operation
+    tool.timeout_sec = 1.0  # Short timeout
 
-    assert tool.args_schema == GetGrippingPointToolInput
+    result = tool._run("test")
+    assert "timed out" in result.lower() or "timeout" in result.lower()

tests/tools/ros2/test_gripping_points.py

@@ -25,6 +25,8 @@
 pytest tests/tools/ros2/test_gripping_points.py::test_gripping_points_manipulation_demo -m "" -s -v
 """
 
+import time
+
 import cv2
 import numpy as np
 import pytest
@@ -278,6 +280,8 @@ def main(config_name: str = "manipulation-demo", test_object: str = "cube"):
         filter_config = algo_config["filter"]
         point_cloud_filter = PointCloudFilter(**filter_config)
 
+        start_time = time.time()
+
         # Create the tool
         gripping_tool = GetGrippingPointTool(
             connector=connector,
@@ -288,7 +292,9 @@ def main(config_name: str = "manipulation-demo", test_object: str = "cube"):
             camera_info_topic=config["topics"]["camera_info"],
             gripping_point_estimator=gripping_estimator,
             point_cloud_filter=point_cloud_filter,
+            timeout_sec=15.0,
         )
+        print(f"elapsed time: {time.time() - start_time} seconds")
 
         # Test the tool directly
         print(f"\nTesting GetGrippingPointTool with object '{test_object}'")
@@ -300,6 +306,8 @@ def main(config_name: str = "manipulation-demo", test_object: str = "cube"):
         for i, gp in enumerate(gripping_points):
             print(f"  GP{i + 1}: [{gp[0]:.3f}, {gp[1]:.3f}, {gp[2]:.3f}]")
 
+        assert len(gripping_points) > 0, "No gripping points found"
+
         if gripping_points:
             # Call the function in pcl.py to publish the gripping point for visualization
             segmented_clouds = gripping_tool.point_cloud_from_segmentation.run(
@@ -319,9 +327,6 @@ def main(config_name: str = "manipulation-demo", test_object: str = "cube"):
             )
             print(f"✅ Saved annotated image as '{annotated_image_path}'")
 
-        else:
-            print("❌ No gripping points found")
-
     except Exception as e:
         print(f"❌ Setup failed: {e}")
         import traceback