Feature: add executor.create_future() (#1495)

* feature: add create_future and test

Signed-off-by: Nadav Elkabets <[email protected]>

* Use create_future in all executor tests

Signed-off-by: Nadav Elkabets <[email protected]>

---------

Signed-off-by: Nadav Elkabets <[email protected]>

Author: nadavelkabets

rclpy/rclpy/executors.py

@@ -260,6 +260,10 @@ def create_task(self, callback: Callable[..., Any], *args: Any, **kwargs: Any
         # Task inherits from Future
         return task
 
+    def create_future(self) -> Future:
+        """Create a Future object attached to the Executor."""
+        return Future(executor=self)
+
     def shutdown(self, timeout_sec: Optional[float] = None) -> bool:
         """
         Stop executing callbacks and wait for their completion.

rclpy/src/rclpy/events_executor/events_executor.cpp

@@ -52,6 +52,7 @@ EventsExecutor::EventsExecutor(py::object context)
   inspect_iscoroutine_(py::module_::import("inspect").attr("iscoroutine")),
   inspect_signature_(py::module_::import("inspect").attr("signature")),
   rclpy_task_(py::module_::import("rclpy.task").attr("Task")),
+  rclpy_future_(py::module_::import("rclpy.task").attr("Future")),
   rclpy_timer_timer_info_(py::module_::import("rclpy.timer").attr("TimerInfo")),
   signal_callback_([this]() {events_queue_.Stop();}),
   rcl_callback_manager_(&events_queue_),
@@ -78,6 +79,12 @@ pybind11::object EventsExecutor::create_task(
   return task;
 }
 
+pybind11::object EventsExecutor::create_future()
+{
+  using py::literals::operator""_a;
+  return rclpy_future_("executor"_a = py::cast(this));
+}
+
 bool EventsExecutor::shutdown(std::optional<double> timeout)
 {
   // NOTE: The rclpy context can invoke this with a lock on the context held.  Therefore we must
@@ -897,6 +904,7 @@ void define_events_executor(py::object module)
   .def(py::init<py::object>(), py::arg("context"))
   .def_property_readonly("context", &EventsExecutor::get_context)
   .def("create_task", &EventsExecutor::create_task, py::arg("callback"))
+  .def("create_future", &EventsExecutor::create_future)
   .def("shutdown", &EventsExecutor::shutdown, py::arg("timeout_sec") = py::none())
   .def("add_node", &EventsExecutor::add_node, py::arg("node"))
   .def("remove_node", &EventsExecutor::remove_node, py::arg("node"))

rclpy/src/rclpy/events_executor/events_executor.hpp

@@ -67,6 +67,7 @@ class EventsExecutor
   pybind11::object get_context() const {return rclpy_context_;}
   pybind11::object create_task(
     pybind11::object callback, pybind11::args args = {}, const pybind11::kwargs & kwargs = {});
+  pybind11::object create_future();
   bool shutdown(std::optional<double> timeout_sec = {});
   bool add_node(pybind11::object node);
   void remove_node(pybind11::handle node);
@@ -168,6 +169,7 @@ class EventsExecutor
   const pybind11::object inspect_iscoroutine_;
   const pybind11::object inspect_signature_;
   const pybind11::object rclpy_task_;
+  const pybind11::object rclpy_future_;
   const pybind11::object rclpy_timer_timer_info_;
 
   EventsQueue events_queue_;

rclpy/test/test_executor.py

@@ -494,7 +494,7 @@ def timer_callback() -> None:
                 timer = self.node.create_timer(0.003, timer_callback)
 
                 # Timeout
-                future = Future[None]()
+                future = executor.create_future()
                 self.assertFalse(future.done())
                 start = time.perf_counter()
                 executor.spin_until_future_complete(future=future, timeout_sec=0.1)
@@ -521,7 +521,7 @@ def set_future_result(future: Future[str]) -> None:
                     future.set_result('finished')
 
                 # Future complete timeout_sec > 0
-                future = Future[str]()
+                future = executor.create_future()
                 self.assertFalse(future.done())
                 t = threading.Thread(target=lambda: set_future_result(future))
                 t.start()
@@ -530,7 +530,7 @@ def set_future_result(future: Future[str]) -> None:
                 self.assertEqual(future.result(), 'finished')
 
                 # Future complete timeout_sec = None
-                future = Future()
+                future = executor.create_future()
                 self.assertFalse(future.done())
                 t = threading.Thread(target=lambda: set_future_result(future))
                 t.start()
@@ -539,7 +539,7 @@ def set_future_result(future: Future[str]) -> None:
                 self.assertEqual(future.result(), 'finished')
 
                 # Future complete timeout < 0
-                future = Future()
+                future = executor.create_future()
                 self.assertFalse(future.done())
                 t = threading.Thread(target=lambda: set_future_result(future))
                 t.start()
@@ -561,7 +561,7 @@ def timer_callback() -> None:
                 timer = self.node.create_timer(0.003, timer_callback)
 
                 # Do not wait timeout_sec = 0
-                future = Future[None]()
+                future = executor.create_future()
                 self.assertFalse(future.done())
                 executor.spin_until_future_complete(future=future, timeout_sec=0)
                 self.assertFalse(future.done())
@@ -644,7 +644,7 @@ def test_single_threaded_spin_once_until_future(self) -> None:
             with self.subTest(cls=cls):
                 executor = cls(context=self.context)
 
-                future = Future[bool](executor=executor)
+                future = executor.create_future()
 
                 # Setup a thread to spin_once_until_future_complete, which will spin
                 # for a maximum of 10 seconds.
@@ -672,7 +672,7 @@ def test_multi_threaded_spin_once_until_future(self) -> None:
         self.assertIsNotNone(self.node.handle)
         executor = MultiThreadedExecutor(context=self.context)
 
-        future: Future[bool] = Future(executor=executor)
+        future: Future[bool] = executor.create_future()
 
         # Setup a thread to spin_once_until_future_complete, which will spin
         # for a maximum of 10 seconds.
@@ -721,7 +721,7 @@ def timer2_callback() -> None:
                 timer2 = self.node.create_timer(1.5, timer2_callback, callback_group)
 
                 executor.add_node(self.node)
-                future = Future[None](executor=executor)
+                future = executor.create_future()
                 executor.spin_until_future_complete(future, 4)
 
                 assert count == 2
@@ -731,6 +731,17 @@ def timer2_callback() -> None:
                 self.node.destroy_timer(timer1)
                 self.node.destroy_client(cli)
 
+    def test_create_future_returns_future_with_executor_attached(self) -> None:
+        self.assertIsNotNone(self.node.handle)
+        for cls in [SingleThreadedExecutor, MultiThreadedExecutor, EventsExecutor]:
+            with self.subTest(cls=cls):
+                executor = cls(context=self.context)
+                try:
+                    fut = executor.create_future()
+                    self.assertEqual(executor, fut._executor())
+                finally:
+                    executor.shutdown()
+
 
 if __name__ == '__main__':
     unittest.main()