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 the Epoll class.
- Introduce `EpollEventSource` as an abstract base class for specific event handlers.
- Replace raw file descriptor `epoll_fd_` usage with the encapsulated `Epoll` class.
- Move implementation details from headers to `.cpp` files.

Relates to: #969

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

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_epoll_update_dispatcher.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_event.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.hpp"
 #include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_smart_pointer.hpp"
 
@@ -9,6 +10,8 @@
 namespace agnocast
 {
 
+constexpr uint32_t MAX_CALLBACK_INFO_ID = 0x10000000;
+
 struct AgnocastExecutable;
 
 // Base class for a type-erased object
@@ -124,4 +127,28 @@ void enqueue_receive_and_execute(
   std::mutex & ready_agnocast_executables_mutex,
   std::vector<AgnocastExecutable> & ready_agnocast_executables);
 
+class SubscriptionEventSource : public EpollEventSource
+{
+  pid_t my_pid_;
+  std::mutex * ready_agnocast_executables_mutex_;
+  std::vector<AgnocastExecutable> * ready_agnocast_executables_;
+
+public:
+  SubscriptionEventSource(
+    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(
+    Epoll & epoll, const CallbackGroupValidator & validate_callback_group) override;
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
 }  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_epoll.hpp

@@ -1,8 +1,8 @@
 #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 <rclcpp/callback_group.hpp>
 
 #include <atomic>
 #include <mutex>
@@ -12,108 +12,67 @@
 namespace agnocast
 {
 
-struct AgnocastExecutable;
+using CallbackGroupValidator = std::function<bool(const rclcpp::CallbackGroup::SharedPtr &)>;
+
+class EpollEventSource
+{
+public:
+  EpollEventSource() = default;
+
+  virtual ~EpollEventSource() = default;
 
-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;
+  EpollEventSource(const EpollEventSource &) = delete;
+  EpollEventSource & operator=(const EpollEventSource &) = delete;
 
-// @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);
+  EpollEventSource(EpollEventSource &&) = delete;
+  EpollEventSource & operator=(EpollEventSource &&) = delete;
 
-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)
+  [[nodiscard]] virtual EpollEventType get_type() const = 0;
+
+  virtual void prepare_epoll(
+    Epoll & epoll, const CallbackGroupValidator & validate_callback_group) = 0;
+
+  virtual bool handle(EpollEventLocalID event_local_id) = 0;
+};
+
+// Shutdown event - only used by AgnocastOnlyExecutor
+class ShutdownEventSource : public EpollEventSource
 {
-  // Register subscription callbacks to epoll
+public:
+  ShutdownEventSource() = default;
+
+  [[nodiscard]] EpollEventType get_type() const override { return EpollEventType::Shutdown; }
+
+  void prepare_epoll(Epoll & epoll, 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;
+    (void)validate_callback_group;
   }
 
-  // Register timers to epoll
+  bool handle(EpollEventLocalID /*event_local_id*/) override { return true; }
+};
+
+using EventSourceArray =
+  std::array<std::unique_ptr<EpollEventSource>, static_cast<size_t>(EpollEventType::NrEventType)>;
+
+class EpollManager
+{
+public:
+  explicit EpollManager(EventSourceArray sources);
+
+  int add_event(int fd, EpollEventType type, EpollEventLocalID local_id)
   {
-    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;
-    }
+    return epoll_.add_source(fd, type, local_id);
   }
-}
+
+  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:
+  Epoll epoll_;
+  EventSourceArray sources_{nullptr};
+};
 
 }  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_epoll_event.hpp

@@ -0,0 +1,62 @@
+#pragma once
+
+#include <cstdint>
+#include <vector>
+
+namespace agnocast
+{
+
+enum class EpollEventType : uint32_t {
+  Subscription = 0,
+  Timer = 1,
+  Shutdown = 2,
+  NrEventType,
+};
+
+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};
+}
+
+using EpollResult = std::pair<EpollEventType, EpollEventLocalID>;
+
+class Epoll
+{
+  static constexpr int EPOLL_WAIT_MAX_EVENTS = 1;
+  int epoll_fd_;
+
+public:
+  Epoll();
+
+  Epoll(const Epoll &) = delete;
+  Epoll & operator=(const Epoll &) = delete;
+
+  Epoll(Epoll && other) noexcept : epoll_fd_(other.epoll_fd_) { other.epoll_fd_ = -1; }
+  Epoll & operator=(Epoll && other) noexcept;
+
+  ~Epoll();
+
+  [[nodiscard]] int add_source(int fd, EpollEventType type, EpollEventLocalID local_id) const;
+
+  void remove(int fd) const;
+
+  int wait(std::vector<EpollResult> & results, int timeout_ms) const;
+
+  [[nodiscard]] int fd() const { return epoll_fd_; }
+
+  [[nodiscard]] bool is_valid() const { return epoll_fd_ >= 0; }
+};
+
+}  // namespace agnocast

src/agnocastlib/include/agnocast/agnocast_executor.hpp

@@ -1,9 +1,13 @@
 #pragma once
 
 #include "agnocast/agnocast_epoll.hpp"
+#include "agnocast/agnocast_epoll_event.hpp"
+#include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_public_api.hpp"
 #include "rclcpp/rclcpp.hpp"
 
+#include <memory>
+
 namespace agnocast
 {
 
@@ -33,7 +37,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,5 +1,6 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll.hpp"
 #include "agnocast/agnocast_timer.hpp"
 #include "rclcpp/rclcpp.hpp"
 
@@ -15,6 +16,9 @@ namespace agnocast
 {
 
 constexpr int64_t NANOSECONDS_PER_SECOND = 1000000000;
+constexpr uint32_t MAX_TIMER_ID = 0x10000000;
+
+struct AgnocastExecutable;
 
 struct TimerInfo
 {
@@ -59,4 +63,28 @@ void register_timer_info(
 
 void unregister_timer_info(uint32_t timer_id);
 
+class TimerEventSource : public EpollEventSource
+{
+  pid_t my_pid_;
+  std::mutex * ready_agnocast_executables_mutex_;
+  std::vector<AgnocastExecutable> * ready_agnocast_executables_;
+
+public:
+  TimerEventSource(
+    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(
+    Epoll & epoll, const CallbackGroupValidator & validate_callback_group) override;
+
+  bool handle(EpollEventLocalID event_local_id) override;
+};
+
 }  // namespace agnocast

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

@@ -1,5 +1,7 @@
 #pragma once
 
+#include "agnocast/agnocast_epoll.hpp"
+#include "agnocast/agnocast_epoll_event.hpp"
 #include "agnocast/agnocast_epoll_update_dispatcher.hpp"
 #include "agnocast/agnocast_public_api.hpp"
 #include "rclcpp/callback_group.hpp"
@@ -30,7 +32,7 @@ class AgnocastOnlyExecutor
 {
 protected:
   std::atomic_bool spinning_;
-  int epoll_fd_;
+  std::unique_ptr<EpollManager> epoll_manager_;
   int shutdown_event_fd_;
   pid_t my_pid_;