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

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

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().notify_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,70 @@
+#pragma once
+
+#include "sys/epoll.h"
+
+#include <atomic>
+#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 notify_all();
+
+  EpollUpdateTracker create_tracker();
+
+private:
+  EpollUpdateDispatcher() = default;
+  friend class EpollUpdateTracker;
+
+  void unregister(int tracker_id);
+
+  std::atomic<int> next_tracker_id_{1};
+
+  std::mutex mutex_;
+  std::unordered_map<int, std::shared_ptr<TrackerContext>> trackers_;
+};
+
+class EpollUpdateTracker
+{
+public:
+  EpollUpdateTracker(const EpollUpdateTracker &) = delete;
+  EpollUpdateTracker & operator=(const EpollUpdateTracker &) = delete;
+
+  EpollUpdateTracker(EpollUpdateTracker && other) noexcept;
+  EpollUpdateTracker & operator=(EpollUpdateTracker && other) noexcept;
+
+  ~EpollUpdateTracker();
+
+  [[nodiscard]] bool need_update() const;
+
+private:
+  friend class EpollUpdateDispatcher;
+
+  EpollUpdateTracker(int tracker_id, std::shared_ptr<TrackerContext> context)
+  : id_(tracker_id), context_(std::move(context))
+  {
+  }
+
+  int id_;
+  std::shared_ptr<TrackerContext> context_;
+};
+
+}  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_executor.hpp

@@ -41,6 +41,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,71 @@
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
+
+#include <atomic>
+
+namespace agnocast
+{
+
+EpollUpdateTracker EpollUpdateDispatcher::create_tracker()
+{
+  int 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::notify_all()
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  for (auto & [id, context] : trackers_) {
+    context->need_update.store(true, std::memory_order_release);
+  }
+}
+
+void EpollUpdateDispatcher::unregister(int tracker_id)
+{
+  std::lock_guard<std::mutex> lock(mutex_);
+  trackers_.erase(tracker_id);
+}
+
+EpollUpdateTracker::EpollUpdateTracker(EpollUpdateTracker && other) noexcept
+: id_(other.id_), context_(std::move(other.context_))
+{
+  other.id_ = 0;
+}
+
+EpollUpdateTracker & EpollUpdateTracker::operator=(EpollUpdateTracker && other) noexcept
+{
+  if (this != &other) {
+    if (id_ != 0) {
+      EpollUpdateDispatcher::get_instance().unregister(id_);
+    }
+    id_ = other.id_;
+    context_ = std::move(other.context_);
+
+    other.id_ = 0;
+  }
+  return *this;
+}
+
+EpollUpdateTracker::~EpollUpdateTracker()
+{
+  if (id_ != 0) {
+    EpollUpdateDispatcher::get_instance().unregister(id_);
+  }
+}
+
+bool EpollUpdateTracker::need_update() const
+{
+  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().create_tracker())
 {
   if (epoll_fd_ == -1) {
     RCLCPP_ERROR(logger, "epoll_create1 failed: %s", strerror(errno));

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_.need_update()) {
       prepare_epoll();
     }