refactor(agnocastlib): extract event-specific logic from agnocast_epoll

Previously, all event-specific logic was hardcoded in `agnocast_epoll.cpp`
and `.hpp`. This caused dependency issues and made it difficult to add
new event types.

This commit refactors the code by categorizing events and moving their
implementations into separate source files. Key changes include:

- Change epoll_data format from u32 to u64 to hold both event kind and local identifier.
- Introduce `EpollManager` to manage and dispatch events from epoll.
- Introduce `EpollEventHandler` as an abstract base class for specific event handlers.

Relates to: #969

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

Author: ruth561

src/agnocastlib/include/agnocast/agnocast_callback_info.hpp

@@ -1,14 +1,21 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll.hpp"
 #include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_smart_pointer.hpp"
 
+#include <cstdint>
 #include <mutex>
 #include <type_traits>
 
 namespace agnocast
 {
 
+// Capped slightly below UINT32_MAX to provide a safe margin against
+// atomic wrap-around (overflow back to 0) during concurrent fetch_add calls.
+constexpr uint32_t MAX_CALLBACK_INFO_ID_SAFETY_MARGIN = 1000;
+constexpr uint32_t MAX_CALLBACK_INFO_ID = UINT32_MAX - MAX_CALLBACK_INFO_ID_SAFETY_MARGIN;
+
 struct AgnocastExecutable;
 
 // Base class for a type-erased object
@@ -124,4 +131,27 @@ void enqueue_receive_and_execute(
   std::mutex & ready_agnocast_executables_mutex,
   std::vector<AgnocastExecutable> & ready_agnocast_executables);
 
+class SubscriptionEventHandler : public EpollEventHandler
+{
+  pid_t my_pid_;
+  std::mutex * ready_agnocast_executables_mutex_;
+  std::vector<AgnocastExecutable> * ready_agnocast_executables_;
+
+public:
+  SubscriptionEventHandler(
+    const pid_t my_pid, std::mutex * ready_agnocast_executables_mutex,
+    std::vector<AgnocastExecutable> * ready_agnocast_executables)
+  : my_pid_(my_pid),
+    ready_agnocast_executables_mutex_(ready_agnocast_executables_mutex),
+    ready_agnocast_executables_(ready_agnocast_executables)
+  {
+  }
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Subscription; }
+
+  void prepare_epoll(int epoll_fd, const CallbackGroupValidator & validate_callback_group) override;
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
 }  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_epoll.hpp

@@ -1,119 +1,95 @@
 #pragma once
 
-#include "agnocast/agnocast_callback_info.hpp"
-#include "agnocast/agnocast_timer_info.hpp"
-#include "sys/epoll.h"
+#include "agnocast/agnocast_epoll_event.hpp"
 
-#include <atomic>
-#include <mutex>
-#include <shared_mutex>
-#include <vector>
+#include <rclcpp/callback_group.hpp>
+
+#include <array>
+#include <functional>
+#include <memory>
 
 namespace agnocast
 {
 
-struct AgnocastExecutable;
+using CallbackGroupValidator = std::function<bool(const rclcpp::CallbackGroup::SharedPtr &)>;
+
+class EpollEventHandler
+{
+public:
+  EpollEventHandler() = default;
+
+  virtual ~EpollEventHandler() = default;
+
+  EpollEventHandler(const EpollEventHandler &) = delete;
+  EpollEventHandler & operator=(const EpollEventHandler &) = delete;
+
+  EpollEventHandler(EpollEventHandler &&) = delete;
+  EpollEventHandler & operator=(EpollEventHandler &&) = delete;
 
-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)
-constexpr uint32_t EPOLL_EVENT_ID_RESERVED_MASK =
-  TIMER_EVENT_FLAG | CLOCK_EVENT_FLAG | SHUTDOWN_EVENT_FLAG;
+  [[nodiscard]] virtual EpollEventType get_type() const = 0;
 
-// @return true if shutdown event detected, false otherwise
-bool wait_and_handle_epoll_event(
-  int epoll_fd, pid_t my_pid, int timeout_ms, std::mutex & ready_agnocast_executables_mutex,
-  std::vector<AgnocastExecutable> & ready_agnocast_executables);
+  virtual void prepare_epoll(
+    int epoll_fd, const CallbackGroupValidator & validate_callback_group) = 0;
 
-template <class ValidateFn>
-void prepare_epoll_impl(
-  const int epoll_fd, const pid_t my_pid, std::mutex & ready_agnocast_executables_mutex,
-  std::vector<AgnocastExecutable> & ready_agnocast_executables,
-  ValidateFn && validate_callback_group)
+  virtual bool handle(EpollEventLocalID event_local_id) = 0;
+};
+
+// Shutdown event - only used by AgnocastOnlyExecutor
+class ShutdownEventHandler : public EpollEventHandler
 {
-  // Register subscription callbacks to epoll
+public:
+  ShutdownEventHandler() = default;
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Shutdown; }
+
+  void prepare_epoll(int epoll_fd, const CallbackGroupValidator & validate_callback_group) override
   {
-    std::lock_guard<std::mutex> lock(id2_callback_info_mtx);
-
-    for (auto & it : id2_callback_info) {
-      const uint32_t callback_info_id = it.first;
-      CallbackInfo & callback_info = it.second;
-
-      if (!callback_info.need_epoll_update) {
-        continue;
-      }
-
-      if (!validate_callback_group(callback_info.callback_group)) {
-        continue;
-      }
-
-      struct epoll_event ev = {};
-      ev.events = EPOLLIN;
-      ev.data.u32 = callback_info_id;
-
-      if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, callback_info.mqdes, &ev) == -1) {
-        RCLCPP_ERROR(logger, "epoll_ctl failed: %s", strerror(errno));
-        close(agnocast_fd);
-        exit(EXIT_FAILURE);
-      }
-
-      if (callback_info.is_transient_local) {
-        agnocast::enqueue_receive_and_execute(
-          callback_info_id, my_pid, callback_info, ready_agnocast_executables_mutex,
-          ready_agnocast_executables);
-      }
-
-      callback_info.need_epoll_update = false;
-    }
+    (void)epoll_fd;
+    (void)validate_callback_group;
   }
 
-  // Register timers to epoll
+  bool handle(EpollEventLocalID /*event_local_id*/) override { return true; }
+};
+
+class DummyEventHandler : public EpollEventHandler
+{
+public:
+  DummyEventHandler() = default;
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Dummy; }
+
+  void prepare_epoll(int epoll_fd, const CallbackGroupValidator & validate_callback_group) override
   {
-    std::lock_guard<std::mutex> lock(id2_timer_info_mtx);
-
-    for (auto & it : id2_timer_info) {
-      const uint32_t timer_id = it.first;
-      TimerInfo & timer_info = *it.second;
-
-      if (!timer_info.need_epoll_update) {
-        continue;
-      }
-
-      if (!timer_info.timer.lock() || !validate_callback_group(timer_info.callback_group)) {
-        continue;
-      }
-
-      std::shared_lock fd_lock(timer_info.fd_mutex);
-
-      // Register clock_eventfd for ROS_TIME timers (simulation time support)
-      if (timer_info.clock_eventfd >= 0) {
-        struct epoll_event clock_ev = {};
-        clock_ev.events = EPOLLIN;
-        clock_ev.data.u32 = timer_id | CLOCK_EVENT_FLAG;
-
-        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, timer_info.clock_eventfd, &clock_ev) == -1) {
-          RCLCPP_ERROR(logger, "epoll_ctl failed for clock_eventfd: %s", strerror(errno));
-          close(agnocast_fd);
-          exit(EXIT_FAILURE);
-        }
-      }
-
-      // Register timerfd (wall clock based firing)
-      if (timer_info.timer_fd >= 0) {
-        struct epoll_event ev = {};
-        ev.events = EPOLLIN;
-        ev.data.u32 = timer_id | TIMER_EVENT_FLAG;
-
-        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, timer_info.timer_fd, &ev) == -1) {
-          RCLCPP_ERROR(logger, "epoll_ctl failed for timer: %s", strerror(errno));
-          close(agnocast_fd);
-          exit(EXIT_FAILURE);
-        }
-      }
-
-      timer_info.need_epoll_update = false;
-    }
+    (void)epoll_fd;
+    (void)validate_callback_group;
   }
-}
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
+using EventHandlerArray =
+  std::array<std::unique_ptr<EpollEventHandler>, static_cast<size_t>(EpollEventType::NrEventType)>;
+
+class EpollManager final
+{
+public:
+  explicit EpollManager(EventHandlerArray sources);
+  ~EpollManager();
+  EpollManager(const EpollManager &) = delete;
+  EpollManager & operator=(const EpollManager &) = delete;
+  EpollManager(EpollManager &&) = delete;
+  EpollManager & operator=(EpollManager &&) = delete;
+
+  [[nodiscard]] int add_event(int fd, EpollEventType type, EpollEventLocalID local_id) const;
+
+  void prepare_epoll(const CallbackGroupValidator & validate_callback_group);
+
+  /// @return true if shutdown event detected, false otherwise
+  bool wait_and_handle_epoll_event(int timeout_ms);
+
+private:
+  int epoll_fd_{-1};
+  EventHandlerArray sources_{nullptr};
+};
 
 }  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_epoll_event.hpp

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <cstdint>
+#include <utility>
+
+namespace agnocast
+{
+
+enum class EpollEventType : uint32_t {
+  Subscription = 0,
+  Timer,
+  Clock,
+  Shutdown,
+  NrEventType,
+
+  Dummy = 0xFFFFFFFF,
+};
+
+using EpollEventLocalID = uint32_t;
+
+constexpr uint32_t EPOLL_DATA_TYPE_SHIFT = 32;
+constexpr uint32_t EPOLL_DATA_ID_BITMASK = 0xFFFFFFFF;
+
+inline uint64_t pack_epoll_data(EpollEventType type, EpollEventLocalID id)
+{
+  return (static_cast<uint64_t>(type) << EPOLL_DATA_TYPE_SHIFT) | id;
+}
+
+inline std::pair<EpollEventType, EpollEventLocalID> unpack_epoll_data(uint64_t data)
+{
+  auto type = static_cast<EpollEventType>(data >> EPOLL_DATA_TYPE_SHIFT);
+  auto id = static_cast<uint32_t>(data & EPOLL_DATA_ID_BITMASK);
+  return {type, id};
+}
+
+}  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_executor.hpp

@@ -5,6 +5,8 @@
 #include "agnocast/agnocast_public_api.hpp"
 #include "rclcpp/rclcpp.hpp"
 
+#include <memory>
+
 namespace agnocast
 {
 
@@ -34,7 +36,7 @@ class AgnocastExecutor : public rclcpp::Executor
   virtual bool validate_callback_group(const rclcpp::CallbackGroup::SharedPtr & group) const = 0;
 
 protected:
-  int epoll_fd_;
+  std::unique_ptr<EpollManager> epoll_manager_;
   pid_t my_pid_;
   std::mutex wait_mutex_;
 

src/agnocastlib/include/agnocast/agnocast_timer_info.hpp

@@ -1,10 +1,12 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll.hpp"
 #include "agnocast/agnocast_timer.hpp"
 #include "rclcpp/rclcpp.hpp"
 
 #include <atomic>
 #include <chrono>
+#include <cstdint>
 #include <functional>
 #include <memory>
 #include <mutex>
@@ -15,6 +17,12 @@ namespace agnocast
 {
 
 constexpr int64_t NANOSECONDS_PER_SECOND = 1000000000;
+// Capped slightly below UINT32_MAX to provide a safe margin against
+// atomic wrap-around (overflow back to 0) during concurrent fetch_add calls.
+constexpr uint32_t MAX_TIMER_ID_SAFETY_MARGIN = 1000;
+constexpr uint32_t MAX_TIMER_ID = UINT32_MAX - MAX_TIMER_ID_SAFETY_MARGIN;
+
+struct AgnocastExecutable;
 
 struct TimerInfo
 {
@@ -27,7 +35,9 @@ struct TimerInfo
 
   uint32_t timer_id = 0;
   int timer_fd = -1;
+  bool timer_fd_need_update = false;
   int clock_eventfd = -1;  // eventfd to wake epoll on clock updates (ROS_TIME only)
+  bool clock_eventfd_need_update = false;
   std::weak_ptr<TimerBase> timer;
   rclcpp::CallbackGroup::SharedPtr callback_group;
   std::atomic<int64_t> last_call_time_ns;
@@ -59,4 +69,50 @@ void register_timer_info(
 
 void unregister_timer_info(uint32_t timer_id);
 
+class TimerEventHandler : public EpollEventHandler
+{
+  pid_t my_pid_;
+  std::mutex * ready_agnocast_executables_mutex_;
+  std::vector<AgnocastExecutable> * ready_agnocast_executables_;
+
+public:
+  TimerEventHandler(
+    const pid_t my_pid, std::mutex * ready_agnocast_executables_mutex,
+    std::vector<AgnocastExecutable> * ready_agnocast_executables)
+  : my_pid_(my_pid),
+    ready_agnocast_executables_mutex_(ready_agnocast_executables_mutex),
+    ready_agnocast_executables_(ready_agnocast_executables)
+  {
+  }
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Timer; }
+
+  void prepare_epoll(int epoll_fd, const CallbackGroupValidator & validate_callback_group) override;
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
+class ClockEventHandler : public EpollEventHandler
+{
+  pid_t my_pid_;
+  std::mutex * ready_agnocast_executables_mutex_;
+  std::vector<AgnocastExecutable> * ready_agnocast_executables_;
+
+public:
+  ClockEventHandler(
+    const pid_t my_pid, std::mutex * ready_agnocast_executables_mutex,
+    std::vector<AgnocastExecutable> * ready_agnocast_executables)
+  : my_pid_(my_pid),
+    ready_agnocast_executables_mutex_(ready_agnocast_executables_mutex),
+    ready_agnocast_executables_(ready_agnocast_executables)
+  {
+  }
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Clock; }
+
+  void prepare_epoll(int epoll_fd, const CallbackGroupValidator & validate_callback_group) override;
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
 }  // namespace agnocast