refactor(agnocastlib): remove the global variable need_epoll_updates (#1271)

* refactor(agnocastlib): remove the global variable need_epoll_updates

Previously, when events managed by epoll changed, we notified each Executor
to call `prepare_epoll_impl()` by setting the global atomic variable
`need_epoll_updates` to true. However, this implementation will become
problematic for future refactoring efforts aimed at extracting event-specific
processing from `agnocast_epoll.hpp` and `.cpp`.

This commit removes this global variable and introduces an alternative
notification mechanism. While the current implementation only supports
broadcasting notifications to all Executors—which leaves some performance
challenges—it establishes a 1-to-1 tracking structure for each Executor.
This lays the groundwork for implementing targeted 1-to-1 notifications
in the future.

Relates to: #969

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* test(agnocastlib): Add integration tests for epoll updates

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* test(agnocastlib): Add unit tests for epoll updates

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* refactor(agnocastlib): reorganize header includes

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* refactor(agnocastlib): remove epoll_update_tracker_id_ from AgnocastOnlyExecutor

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* fix(agnocastlib): fix segfault by moving stop_spinning to test body

This commit fixes a segmentation fault in the tests.

[Problem]
Previously, the Executor was stopped in `TearDown()`. Since `TearDown()`
runs after the test fixture is destroyed, shared variables `callback_started`
referenced by timer callbacks were already deleted. When the Executor tried
to run a callback using these deleted variables, it caused a crash.

[Solution]
We moved the Executor clean-up (`stop_spinning`) from `TearDown()` into
the test body. This ensures that the Executor stops completely before
the test variables are destroyed.

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

* fix(executor): explicitly add callback groups when building for Humble version

When building `MultiThreadedAgnocastExecutor` for the ROS 2 Humble version,
callback groups created with  `automatically_add_to_executor_with_node==true`
after a node has been associated with an executor are not automatically picked up.

This commit explicitly calls `add_callback_groups_from_nodes_associated_to_executor()`
to fix this behavior, matching the automatic handling within
`get_all_callback_groups()` in the Jazzy version.

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>

---------

Signed-off-by: Takumi Jin <primenumber_2_3_5@yahoo.co.jp>
Co-authored-by: Koichi Imai <45482193+Koichi98@users.noreply.github.com>

Author: ruth561

src/agnocastlib/CMakeLists.txt

@@ -32,7 +32,7 @@ find_package(glog REQUIRED)
 
 add_library(agnocast SHARED
   src/agnocast.cpp src/agnocast_utils.cpp src/agnocast_publisher.cpp src/agnocast_subscription.cpp
-  src/agnocast_smart_pointer.cpp src/agnocast_callback_info.cpp src/agnocast_epoll.cpp src/agnocast_timer_info.cpp src/agnocast_timer.cpp src/agnocast_executor.cpp
+  src/agnocast_smart_pointer.cpp src/agnocast_callback_info.cpp src/agnocast_epoll.cpp src/agnocast_epoll_update_dispatcher.cpp src/agnocast_timer_info.cpp src/agnocast_timer.cpp src/agnocast_executor.cpp
   src/agnocast_single_threaded_executor.cpp src/agnocast_multi_threaded_executor.cpp
   src/agnocast_callback_isolated_executor.cpp
   src/agnocast_tracepoint_wrapper.c src/agnocast_client.cpp
@@ -130,6 +130,7 @@ if(BUILD_TESTING)
     test/unit/test_agnocast_subscription.cpp
     test/unit/test_mocked_agnocast.cpp
     test/unit/test_agnocast_executors.cpp
+    test/unit/test_agnocast_epoll_update.cpp
     test/unit/test_node_topics.cpp
     test/unit/test_cie_client_utils.cpp)
   target_include_directories(test_unit_${PROJECT_NAME} PRIVATE include)
@@ -194,6 +195,18 @@ if(BUILD_TESTING)
     LABELS "requires_kernel_module;requires_heaphook"
   )
 
+  # Epoll update integration test (requires kernel module, no mock)
+  ament_add_gmock(test_integration_agnocast_epoll_update_${PROJECT_NAME}
+    test/integration/test_agnocast_epoll_update.cpp)
+  target_link_libraries(test_integration_agnocast_epoll_update_${PROJECT_NAME} agnocast)
+  ament_target_dependencies(test_integration_agnocast_epoll_update_${PROJECT_NAME}
+    agnocast_cie_config_msgs)
+  set_tests_properties(test_integration_agnocast_epoll_update_${PROJECT_NAME} PROPERTIES
+    ENVIRONMENT "GTEST_DEATH_TEST_STYLE=threadsafe;LD_PRELOAD=${CMAKE_INSTALL_PREFIX}/lib/libagnocast_heaphook.so:$ENV{LD_PRELOAD}"
+    TIMEOUT 120
+    LABELS "requires_kernel_module;requires_heaphook"
+  )
+
   # Integration tests for agnocast_heaphook
   add_library(initialize_shutdown_mock SHARED test/integration/src/initialize_shutdown_mock.cpp)
   set_target_properties(initialize_shutdown_mock PROPERTIES

src/agnocastlib/include/agnocast/agnocast_callback_info.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_smart_pointer.hpp"
 
 #include <mutex>
@@ -57,7 +58,6 @@ std::vector<std::string> get_agnocast_topics_by_group(
 extern std::mutex id2_callback_info_mtx;
 extern std::unordered_map<uint32_t, CallbackInfo> id2_callback_info;
 extern std::atomic<uint32_t> next_callback_info_id;
-extern std::atomic<bool> need_epoll_updates;
 
 uint32_t allocate_callback_info_id();
 
@@ -109,7 +109,7 @@ uint32_t register_callback(
                    callback_group, erased_callback, message_creator};
   }
 
-  need_epoll_updates.store(true);
+  EpollUpdateDispatcher::get_instance().request_update_all();
 
   return callback_info_id;
 }

src/agnocastlib/include/agnocast/agnocast_epoll.hpp

@@ -14,8 +14,6 @@ namespace agnocast
 
 struct AgnocastExecutable;
 
-extern std::atomic<bool> need_epoll_updates;
-
 constexpr uint32_t TIMER_EVENT_FLAG = 0x80000000;
 constexpr uint32_t CLOCK_EVENT_FLAG = 0x40000000;     // For clock_eventfd events (ROS_TIME timers)
 constexpr uint32_t SHUTDOWN_EVENT_FLAG = 0x20000000;  // For shutdown events (AgnocastOnlyExecutor)
@@ -116,28 +114,6 @@ void prepare_epoll_impl(
       timer_info.need_epoll_update = false;
     }
   }
-
-  // Check if all updates are done. Both locks must be held simultaneously to prevent
-  // a TOCTOU race: without this, a new subscription could set need_epoll_updates=true
-  // between the two checks (or between the last check and the store), and that update
-  // would be lost when need_epoll_updates is overwritten to false.
-  // Lock ordering: id2_callback_info_mtx before id2_timer_info_mtx (see declarations).
-  {
-    std::lock_guard<std::mutex> cb_lock(id2_callback_info_mtx);
-    std::lock_guard<std::mutex> timer_lock(id2_timer_info_mtx);
-
-    const bool all_callbacks_updated = std::none_of(
-      id2_callback_info.begin(), id2_callback_info.end(),
-      [](const auto & it) { return it.second.need_epoll_update; });
-
-    const bool all_timers_updated = std::none_of(
-      id2_timer_info.begin(), id2_timer_info.end(),
-      [](const auto & it) { return it.second->need_epoll_update; });
-
-    if (all_callbacks_updated && all_timers_updated) {
-      need_epoll_updates.store(false);
-    }
-  }
 }
 
 }  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_epoll_update_dispatcher.hpp

@@ -0,0 +1,71 @@
+#pragma once
+
+#include <atomic>
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+
+namespace agnocast
+{
+
+class EpollUpdateTracker;
+
+struct TrackerContext
+{
+  std::atomic<bool> need_update{true};
+};
+
+class EpollUpdateDispatcher
+{
+public:
+  static EpollUpdateDispatcher & get_instance()
+  {
+    static EpollUpdateDispatcher instance;
+    return instance;
+  }
+
+  void request_update_all();
+
+  void request_update(uint64_t tracker_id);
+
+  EpollUpdateTracker register_tracker();
+
+private:
+  EpollUpdateDispatcher() = default;
+  friend class EpollUpdateTracker;
+
+  void unregister_tracker(uint64_t tracker_id);
+
+  std::atomic<uint64_t> next_tracker_id_{1};
+
+  std::mutex mutex_;
+  std::unordered_map<uint64_t, std::shared_ptr<TrackerContext>> trackers_;
+};
+
+class EpollUpdateTracker
+{
+public:
+  EpollUpdateTracker(const EpollUpdateTracker &) = delete;
+  EpollUpdateTracker & operator=(const EpollUpdateTracker &) = delete;
+  EpollUpdateTracker(EpollUpdateTracker && other) = delete;
+  EpollUpdateTracker & operator=(EpollUpdateTracker && other) = delete;
+  ~EpollUpdateTracker();
+
+  [[nodiscard]] bool take_update_request();
+
+  [[nodiscard]] uint64_t id() const { return id_; }
+
+private:
+  friend class EpollUpdateDispatcher;
+
+  EpollUpdateTracker(uint64_t tracker_id, std::shared_ptr<TrackerContext> context)
+  : id_(tracker_id), context_(std::move(context))
+  {
+  }
+
+  uint64_t id_;
+  std::shared_ptr<TrackerContext> context_;
+};
+
+}  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_executor.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "agnocast/agnocast_epoll.hpp"
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_public_api.hpp"
 #include "rclcpp/rclcpp.hpp"
 
@@ -41,6 +42,8 @@ class AgnocastExecutor : public rclcpp::Executor
   bool get_next_agnocast_executable(AgnocastExecutable & agnocast_executable, const int timeout_ms);
   static void execute_agnocast_executable(AgnocastExecutable & agnocast_executable);
 
+  EpollUpdateTracker epoll_update_tracker_;
+
 public:
   /// Construct the executor.
   /// @param options Executor options.

src/agnocastlib/include/agnocast/node/agnocast_only_executor.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_public_api.hpp"
 #include "rclcpp/callback_group.hpp"
 #include "rclcpp/node_interfaces/node_base_interface.hpp"
@@ -33,6 +34,8 @@ class AgnocastOnlyExecutor
   int shutdown_event_fd_;
   pid_t my_pid_;
 
+  EpollUpdateTracker epoll_update_tracker_;
+
   // Lock ordering: When both mutexes are needed, always acquire
   // ready_agnocast_executables_mutex_ before mutex_ to prevent deadlocks.
   std::mutex ready_agnocast_executables_mutex_;

src/agnocastlib/src/agnocast_epoll.cpp

@@ -7,8 +7,6 @@
 namespace agnocast
 {
 
-std::atomic<bool> need_epoll_updates{false};
-
 bool wait_and_handle_epoll_event(
   const int epoll_fd, const pid_t my_pid, const int timeout_ms,
   std::mutex & ready_agnocast_executables_mutex,

src/agnocastlib/src/agnocast_epoll_update_dispatcher.cpp

@@ -0,0 +1,61 @@
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
+
+#include <atomic>
+#include <cstdint>
+
+namespace agnocast
+{
+
+EpollUpdateTracker EpollUpdateDispatcher::register_tracker()
+{
+  uint64_t new_id = next_tracker_id_.fetch_add(1, std::memory_order_relaxed);
+
+  auto context = std::make_shared<TrackerContext>();
+
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    trackers_.emplace(new_id, context);
+  }
+
+  return {new_id, context};
+}
+
+void EpollUpdateDispatcher::request_update_all()
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  for (auto & [id, context] : trackers_) {
+    context->need_update.store(true, std::memory_order_release);
+  }
+}
+
+void EpollUpdateDispatcher::request_update(uint64_t tracker_id)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  auto it = trackers_.find(tracker_id);
+  if (it != trackers_.end()) {
+    it->second->need_update.store(true, std::memory_order_release);
+  }
+}
+
+void EpollUpdateDispatcher::unregister_tracker(uint64_t tracker_id)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  trackers_.erase(tracker_id);
+}
+
+EpollUpdateTracker::~EpollUpdateTracker()
+{
+  if (id_ != 0) {
+    EpollUpdateDispatcher::get_instance().unregister_tracker(id_);
+  }
+}
+
+bool EpollUpdateTracker::take_update_request()
+{
+  if (!context_) {
+    return false;
+  }
+  return context_->need_update.exchange(false, std::memory_order_acquire);
+}
+
+}  // namespace agnocast

src/agnocastlib/src/agnocast_executor.cpp

@@ -1,6 +1,7 @@
 #include "agnocast/agnocast_executor.hpp"
 
 #include "agnocast/agnocast.hpp"
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_tracepoint_wrapper.h"
 #include "rclcpp/rclcpp.hpp"
 #include "rclcpp/version.h"
@@ -12,7 +13,10 @@ namespace agnocast
 {
 
 AgnocastExecutor::AgnocastExecutor(const rclcpp::ExecutorOptions & options)
-: rclcpp::Executor(options), epoll_fd_(epoll_create1(0)), my_pid_(getpid())
+: rclcpp::Executor(options),
+  epoll_fd_(epoll_create1(0)),
+  my_pid_(getpid()),
+  epoll_update_tracker_(EpollUpdateDispatcher::get_instance().register_tracker())
 {
   if (epoll_fd_ == -1) {
     RCLCPP_ERROR(logger, "epoll_create1 failed: %s", strerror(errno));
@@ -79,6 +83,15 @@ bool AgnocastExecutor::get_next_ready_agnocast_executable(AgnocastExecutable & a
       continue;
     }
 #else
+    // We need to call add_callback_groups_from_nodes_associated_to_executor() to handle the
+    // case where a CallbackGroup is created with automatically_add_to_executor_with_node==true
+    // after the Node has been associated with the Executor.
+    // In Jazzy, this process is handled internally within get_all_callback_groups(), so it
+    // works fine, but in Humble, it must be called explicitly.
+    {
+      std::lock_guard<std::mutex> guard(mutex_);
+      add_callback_groups_from_nodes_associated_to_executor();
+    }
     if (
       rclcpp::Executor::get_node_by_group(
         rclcpp::Executor::weak_groups_to_nodes_, it->callback_group) == nullptr) {

src/agnocastlib/src/agnocast_multi_threaded_executor.cpp

@@ -156,7 +156,7 @@ void MultiThreadedAgnocastExecutor::ros2_spin()
 void MultiThreadedAgnocastExecutor::agnocast_spin()
 {
   while (rclcpp::ok(this->context_) && spinning.load()) {
-    if (need_epoll_updates.load()) {
+    if (epoll_update_tracker_.take_update_request()) {
       prepare_epoll();
     }