Feature/thread safe async await

.github/workflows/tsan.yml

@@ -0,0 +1,52 @@
+name: TSAN Race Check
+
+on:
+  workflow_dispatch:
+
+jobs:
+  tsan:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - ros_distro: humble
+            container_image: rostooling/setup-ros-docker:ubuntu-jammy-latest
+          - ros_distro: jazzy
+            container_image: rostooling/setup-ros-docker:ubuntu-noble-latest
+          - ros_distro: kilted
+            container_image: rostooling/setup-ros-docker:ubuntu-noble-latest
+    runs-on: ubuntu-latest
+    container:
+      image: ${{ matrix.container_image }}
+    env:
+      RMW_IMPLEMENTATION: rmw_cyclonedds_cpp
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Install tooling
+        shell: bash
+        run: |
+          apt-get update
+          apt-get install -y --no-install-recommends util-linux
+
+      - name: Build with TSAN
+        shell: bash
+        run: |
+          source /opt/ros/${{ matrix.ros_distro }}/setup.bash
+          MAKEFLAGS="-j6" colcon build \
+            --packages-select icey icey_examples \
+            --cmake-args \
+              -DCMAKE_BUILD_TYPE=RelWithDebInfo \
+              -DICEY_ENABLE_TSAN=ON \
+              -DICEY_ASYNC_AWAIT_THREAD_SAFE=1
+
+      - name: Run TSAN race repro tests
+        shell: bash
+        run: |
+          source /opt/ros/${{ matrix.ros_distro }}/setup.bash
+          source install/setup.bash
+          TSAN_OPTIONS="halt_on_error=1:detect_deadlocks=1:suppressions=${GITHUB_WORKSPACE}/icey/test/tsan_dds.supp" \
+            setarch x86_64 -R \
+            ${GITHUB_WORKSPACE}/build/icey/test_main \
+            --gtest_filter=NodeTasksThreadSafety.TSanRaceRepro*

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## 0.4.0
 
+### Changed 
+
+- Timer callback signature not longer taking size_t as an argument
+- Added CallbackGroup arguments to API
+
 ### Added
 
 - Support for actions
@@ -22,7 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - allowing coroutines without suspension points (i.e. co_await) but only a single co_return
 - Correct result type implementation
 - Missing request cleanup in ServiceClientImpl::our_to_real_req_id_
-- Added missing cancellation of timeout timer in Promise destruction
+- Missing cancellation of timeout timer in Promise destruction
 
 ## 0.3.0 
 

README.md

@@ -10,6 +10,11 @@
   <img src="https://img.shields.io/github/actions/workflow/status/iv461/icey/build.yml?branch=main&job=build_and_test_ros2_humble%20(humble)&label=ROS%202%20Humble&style=for-the-badge" alt="ROS 2 Humble Build Status" />
 </p>
 
+### Updates:
+- v0.4.0: 
+  - Adds support multi-threaded executor, the async/await API is now thread-safe. 
+  - Also adds support for actions.
+
 ICEY is a new client API for modern asynchronous programming in the Robot Operating System (ROS) 2. It uses C++20 coroutines with async/await syntax for service calls and TF lookups. ICEY allows you to model data flows based on streams and promises. These features simplify application code and make asynchronous data flows clearly visible.
 
 ### Problems ICEY solves:
@@ -22,7 +27,7 @@ ICEY is a new client API for modern asynchronous programming in the Robot Operat
 ICEY is fully compatible with the ROS 2 API since it is built on top of rclcpp. This allows for gradual adoption. It supports all major ROS features: parameters, subscriptions, publishers, timers, services, clients, actions and TF. Additionally, ICEY supports lifecycle nodes using a single API.
 ICEY operates smoothly with the message_filters package, using it for synchronization. ICEY is also extensible, as demonstrated by its support for image transport camera subscription/publishers.
 
-ICEY supports ROS 2 Humble and ROS 2 Jazzy.
+ICEY supports ROS 2 Humble/Jazzy/Kilted.
 
 The [icey_examples](icey_examples) package contains many different example nodes, demonstrating the capabilities of ICEY.
 

icey/CMakeLists.txt

@@ -4,6 +4,9 @@ project(icey)
 set(CMAKE_CXX_STANDARD 20)
 set(CMAKE_CXX_EXTENSIONS OFF)
 
+option(ICEY_ASYNC_AWAIT_THREAD_SAFE "Enable synchronization in async/await internals" ON)
+option(ICEY_ENABLE_TSAN "Build tests with ThreadSanitizer instrumentation" OFF)
+
 if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
   add_compile_options(-Wall -Wextra -pedantic -Werror=return-type -Werror=init-self)
 endif()
@@ -25,11 +28,14 @@ include_directories(
 )
 
 ament_auto_add_library(icey 
+  src/icey_async_await.cpp
   src/actions/client.cpp 
   src/actions/server.cpp 
   src/actions/server_goal_handle.cpp  
   )
 
+
+
 if(BUILD_TESTING)
   find_package(ament_cmake_gtest REQUIRED)
 
@@ -50,6 +56,21 @@ if(BUILD_TESTING)
   target_link_libraries(test_main fmt::fmt)
   set_tests_properties(test_main PROPERTIES TIMEOUT 180)
 
+
+
+  if(ICEY_ENABLE_TSAN)
+    if(CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
+      target_compile_options(icey PUBLIC -fsanitize=thread -fno-omit-frame-pointer)
+      target_link_options(icey PUBLIC -fsanitize=thread)
+      target_compile_options(test_main PRIVATE -fsanitize=thread -fno-omit-frame-pointer)
+      target_link_options(test_main PRIVATE -fsanitize=thread)
+      set_tests_properties(test_main PROPERTIES
+        ENVIRONMENT "TSAN_OPTIONS=halt_on_error=1:suppressions=${CMAKE_CURRENT_SOURCE_DIR}/test/tsan_dds.supp")
+    else()
+      message(WARNING "ICEY_ENABLE_TSAN is ON but compiler does not support TSAN flags")
+    endif()
+  endif()
+
 endif()
 
 if(ament_cmake_auto_VERSION EQUAL 2.7.3)

icey/benchmark/src/events_throughput.cpp

@@ -56,7 +56,9 @@ int main(int argc, char** argv) {
 //   auto exec = rclcpp::experimental::executors::EventsExecutor()
 //   exec.add(node);
 //   exec.spin();
-  rclcpp::spin(node);
+  rclcpp::executors::MultiThreadedExecutor exec{rclcpp::ExecutorOptions(), 8};
+  exec.add_node(node->get_node_base_interface());
+  exec.spin();
   rclcpp::shutdown();
   return 0;
 }

icey/benchmark/src/tf_lookup_async_ref.cpp

@@ -35,5 +35,7 @@ int main(int argc, char **argv) {
         }
       });
 
-  rclcpp::spin(node);
+  rclcpp::executors::MultiThreadedExecutor exec{rclcpp::ExecutorOptions(), 8};
+  exec.add_node(node->get_node_base_interface());
+  exec.spin();
 }

icey/doc/source/api_ros.md

@@ -21,9 +21,6 @@ Convenience node wrappers containing the Icey context
 ```{doxygenclass} icey::NodeWithIceyContext
 ```
 
-```{doxygenstruct} icey::TransformBufferImpl
-```
-
 ```{doxygenstruct} icey::NodeBase
 ```
 

icey/doc/source/development.md

@@ -50,6 +50,32 @@ Then, do not use FastDDS since it uses exceptions as part of it's regular (inste
 RMW_IMPLEMENTATION=rmw_cyclonedds_cpp gdb ./build/icey/test_main
 ```
 
+
+## TSAN (ThreadSanitizer)
+
+TSAN is used to detect data races with multi-threaded executors. 
+Build both `icey` and `icey_examples` with TSAN instrumentation:
+
+```sh
+cd <colcon-ws root>
+source /opt/ros/jazzy/setup.bash
+MAKEFLAGS="-j6" colcon build --packages-select icey icey_examples --cmake-args -DCMAKE_BUILD_TYPE=RelWithDebInfo -DICEY_ENABLE_TSAN=ON -DICEY_ASYNC_AWAIT_THREAD_SAFE=1
+```
+
+Run an example binary under TSAN:
+
+```sh
+TSAN_OPTIONS="halt_on_error=1:detect_deadlocks=1:suppressions=/home/ivo/colcon_ws/src/icey/icey/test/tsan_dds.supp" setarch x86_64 -R /home/ivo/colcon_ws/install/icey_examples/lib/icey_examples/service_client_async_await_example
+```
+
+Why these workarounds are needed:
+
+- `setarch x86_64 -R`: disables ASLR for the process. Without this, TSAN can fail early with `unexpected memory mapping` on Ubuntu 24 (https://bugs.launchpad.net/ubuntu/+source/linux/+bug/2056762, https://github.com/google/sanitizers/issues/1716#issuecomment-2010399341)
+- `suppressions=.../tsan_dds.supp`: filters for known DDS races and deadlocks that only cause noise. 
+- `halt_on_error=1`: fail fast 
+- `detect_deadlocks=0`: There is lock inversion issue in the rclcpp_actions currently
+
+By setting `-DICEY_ASYNC_AWAIT_THREAD_SAFE=0` during build, you can verify that indeed TSAN is able to detect the data races. 
 ## Run clang-tidy: 
 
 

icey/doc/source/first_icey_node.md

@@ -67,9 +67,4 @@ ICEY represents ROS primitives such as timers as a `Stream`, an abstraction over
 
 We also do not need to store the timer object anywhere, because the lifetime of entities in ICEY is bound to the lifetime of the node. In ICEY, you do not need to store subscriptions/timers/services as members of the class, ICEY does this bookkeeping for you.
 
-
-```{warning}
-ICEY-nodes can currently only be used with a single-threaded executor.
-```
-
 In the following, we will look more closely into how Subscriptions and Timers follow the `Stream` concept and how this changes the way of asynchronous programming. 

icey/include/icey/action/client_goal_handle.hpp

@@ -23,22 +23,11 @@
 #include "rcl_action/action_client.h"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/time.hpp"
+#include "rclcpp_action/client_goal_handle.hpp"
 #include "rclcpp_action/exceptions.hpp"
 #include "rclcpp_action/types.hpp"
 #include "rclcpp_action/visibility_control.hpp"
 
-// I'm defining this in namespace rclcpp_action so that people don't have to change their code unnecessarily. 
-// I could also just include the rclcpp_action header, but just defining the enum here is faster to compile.
-namespace rclcpp_action {
-/// The possible statuses that an action goal can finish with.
-enum class ResultCode : int8_t {
-  UNKNOWN = action_msgs::msg::GoalStatus::STATUS_UNKNOWN,
-  SUCCEEDED = action_msgs::msg::GoalStatus::STATUS_SUCCEEDED,
-  CANCELED = action_msgs::msg::GoalStatus::STATUS_CANCELED,
-  ABORTED = action_msgs::msg::GoalStatus::STATUS_ABORTED
-};
-}  // namespace rclcpp_action
-
 namespace icey::rclcpp_action {
 
 using GoalUUID = std::array<uint8_t, UUID_SIZE>;
@@ -66,14 +55,7 @@ class ClientGoalHandle {
   RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(ClientGoalHandle)
 
   // A wrapper that defines the result of an action
-  struct WrappedResult {
-    /// The unique identifier of the goal
-    GoalUUID goal_id;
-    /// A status to indicate if the goal was canceled, aborted, or succeeded
-    ResultCode code;
-    /// User defined fields sent back with an action
-    typename ActionT::Result::SharedPtr result;
-  };
+  using WrappedResult = typename ::rclcpp_action::ClientGoalHandle<ActionT>::WrappedResult;
 
   using Feedback = typename ActionT::Feedback;
   using Result = typename ActionT::Result;