Backport of the Events Executor (#1391) (#1435)

* Introduce EventsExecutor implementation (#1389)

Signed-off-by: Brad Martin <[email protected]>
Signed-off-by: Brad Martin <[email protected]>
Signed-off-by: Florian Vahl <[email protected]>
Co-authored-by: Brad Martin <[email protected]>
Co-authored-by: Brad Martin <[email protected]>
Co-authored-by: Janosch Machowinski <[email protected]>

Author: 4 people

rclpy/CMakeLists.txt

@@ -84,6 +84,11 @@ pybind11_add_module(_rclpy_pybind11
   src/rclpy/destroyable.cpp
   src/rclpy/duration.cpp
   src/rclpy/clock_event.cpp
+  src/rclpy/events_executor/delayed_event_thread.cpp
+  src/rclpy/events_executor/events_executor.cpp
+  src/rclpy/events_executor/events_queue.cpp
+  src/rclpy/events_executor/rcl_support.cpp
+  src/rclpy/events_executor/timers_manager.cpp
   src/rclpy/exceptions.cpp
   src/rclpy/graph.cpp
   src/rclpy/guard_condition.cpp
@@ -176,6 +181,7 @@ if(BUILD_TESTING)
       test/test_create_node.py
       test/test_create_while_spinning.py
       test/test_destruction.py
+      test/test_events_executor.py
       test/test_executor.py
       test/test_expand_topic_name.py
       test/test_guard_condition.py

rclpy/rclpy/event_handler.py

@@ -261,7 +261,7 @@ def _default_incompatible_type_callback(event):
             pass
 
         if self.matched:
-            event_handlers.append(QoSEventHandler(
+            event_handlers.append(EventHandler(
                 callback_group=callback_group,
                 callback=self.matched,
                 event_type=QoSSubscriptionEventType.RCL_SUBSCRIPTION_MATCHED,
@@ -372,7 +372,7 @@ def _default_incompatible_type_callback(event):
             pass
 
         if self.matched:
-            event_handlers.append(QoSEventHandler(
+            event_handlers.append(EventHandler(
                 callback_group=callback_group,
                 callback=self.matched,
                 event_type=QoSPublisherEventType.RCL_PUBLISHER_MATCHED,

rclpy/rclpy/experimental/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2024-2025 Brad Martin
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .events_executor import EventsExecutor  # noqa: F401

rclpy/rclpy/experimental/events_executor.py

@@ -0,0 +1,43 @@
+# Copyright 2024-2025 Brad Martin
+# Copyright 2024 Merlin Labs, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import faulthandler
+import typing
+
+import rclpy.executors
+from rclpy.impl.implementation_singleton import rclpy_implementation as _rclpy
+
+
+# Try to look like we inherit from the rclpy Executor for type checking purposes without
+# getting any of the code from the base class.
+def EventsExecutor(*, context: typing.Optional[rclpy.Context] = None) -> rclpy.executors.Executor:
+    if context is None:
+        context = rclpy.get_default_context()
+
+    # For debugging purposes, if anything goes wrong in C++ make sure we also get a
+    # Python backtrace dumped with the crash.
+    faulthandler.enable()
+
+    ex = typing.cast(rclpy.executors.Executor, _rclpy.EventsExecutor(context))
+
+    # rclpy.Executor does this too.  Note, the context itself is smart enough to check
+    # for bound methods, and check whether the instances they're bound to still exist at
+    # callback time, so we don't have to worry about tearing down this stale callback at
+    # destruction time.
+    # TODO(bmartin427) This should really be done inside of the EventsExecutor
+    # implementation itself, but I'm unable to figure out a pybind11 incantation that
+    # allows me to pass this bound method call from C++.
+    context.on_shutdown(ex.wake)
+
+    return ex

rclpy/rclpy/task.py

@@ -16,6 +16,7 @@
 import inspect
 import sys
 import threading
+from typing import Callable
 import warnings
 import weakref
 
@@ -197,6 +198,20 @@ def add_done_callback(self, callback):
         if invoke:
             callback(self)
 
+    def remove_done_callback(self, callback: Callable[['Future'], None]) -> bool:
+        """
+        Remove a previously-added done callback.
+
+        Returns true if the given callback was found and removed.  Always fails if the Future was
+        already complete.
+        """
+        with self._lock:
+            try:
+                self._callbacks.remove(callback)
+            except ValueError:
+                return False
+            return True
+
 
 class Task(Future):
     """

rclpy/src/rclpy/_rclpy_pybind11.cpp

@@ -32,6 +32,7 @@
 #include "duration.hpp"
 #include "clock_event.hpp"
 #include "event_handle.hpp"
+#include "events_executor/events_executor.hpp"
 #include "exceptions.hpp"
 #include "graph.hpp"
 #include "guard_condition.hpp"
@@ -247,4 +248,6 @@ PYBIND11_MODULE(_rclpy_pybind11, m) {
   rclpy::define_signal_handler_api(m);
   rclpy::define_clock_event(m);
   rclpy::define_lifecycle_api(m);
+
+  rclpy::events_executor::define_events_executor(m);
 }

rclpy/src/rclpy/events_executor/delayed_event_thread.cpp

@@ -0,0 +1,72 @@
+// Copyright 2025 Brad Martin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+#include "events_executor/delayed_event_thread.hpp"
+
+#include <utility>
+
+namespace rclpy
+{
+namespace events_executor
+{
+
+DelayedEventThread::DelayedEventThread(EventsQueue * events_queue)
+: events_queue_(events_queue), thread_([this]() {RunThread();})
+{
+}
+
+DelayedEventThread::~DelayedEventThread()
+{
+  {
+    std::unique_lock<std::mutex> lock(mutex_);
+    done_ = true;
+  }
+  cv_.notify_one();
+  thread_.join();
+}
+
+void DelayedEventThread::EnqueueAt(
+  std::chrono::steady_clock::time_point when, std::function<void()> handler)
+{
+  {
+    std::unique_lock<std::mutex> lock(mutex_);
+    when_ = when;
+    handler_ = handler;
+  }
+  cv_.notify_one();
+}
+
+void DelayedEventThread::RunThread()
+{
+  std::unique_lock<std::mutex> lock(mutex_);
+  while (!done_) {
+    if (handler_) {
+      // Make sure we don't dispatch a stale wait if it changes while we're waiting.
+      const auto latched_when = when_;
+      if (
+        (std::cv_status::timeout == cv_.wait_until(lock, latched_when)) && handler_ &&
+        (when_ <= latched_when))
+      {
+        auto handler = std::move(handler_);
+        handler_ = {};
+        events_queue_->Enqueue(std::move(handler));
+      }
+    } else {
+      // Wait indefinitely until we get signaled that there's something worth looking at.
+      cv_.wait(lock);
+    }
+  }
+}
+
+}  // namespace events_executor
+}  // namespace rclpy

rclpy/src/rclpy/events_executor/delayed_event_thread.hpp

@@ -0,0 +1,63 @@
+// Copyright 2025 Brad Martin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#ifndef RCLPY__EVENTS_EXECUTOR__DELAYED_EVENT_THREAD_HPP_
+#define RCLPY__EVENTS_EXECUTOR__DELAYED_EVENT_THREAD_HPP_
+
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <thread>
+
+#include "events_executor/events_queue.hpp"
+
+namespace rclpy
+{
+namespace events_executor
+{
+
+/// This object manages posting an event handler to an EventsQueue after a specified delay.  The
+/// current delay may be changed or canceled at any time.  This is done by way of a self-contained
+/// child thread to perform the wait.
+class DelayedEventThread
+{
+public:
+  /// The pointer is aliased and must live for the lifetime of this object.
+  explicit DelayedEventThread(EventsQueue *);
+  ~DelayedEventThread();
+
+  /// Schedules an event handler to be enqueued at the specified time point.  Replaces any previous
+  /// wait and handler, which will never be dispatched if it has not been already.
+  void EnqueueAt(std::chrono::steady_clock::time_point when, std::function<void()> handler);
+
+  /// Cancels any previously-scheduled handler.
+  void Cancel() {EnqueueAt({}, {});}
+
+private:
+  void RunThread();
+
+  EventsQueue * const events_queue_;
+  std::mutex mutex_;
+  bool done_{};
+  std::condition_variable cv_;
+  std::chrono::steady_clock::time_point when_;
+  std::function<void()> handler_;
+  std::thread thread_;
+};
+
+}  // namespace events_executor
+}  // namespace rclpy
+
+#endif  // RCLPY__EVENTS_EXECUTOR__DELAYED_EVENT_THREAD_HPP_