honor cancel() issued before spin_until_future_complete().

rclcpp/include/rclcpp/executor.hpp

@@ -564,9 +564,20 @@ class Executor
   virtual void
   handle_updated_entities(bool notify);
 
-  /// Spinning state, used to prevent multi threaded calls to spin and to cancel blocking spins.
+  /// Spinning state, used to prevent multi threaded calls to spin.
+  /**
+   * This flag is only set and cleared by the spin functions themselves, so it
+   * stays true until a spin actually returns, even after cancel() was called.
+   */
   std::atomic_bool spinning;
 
+  /// Tracks a pending cancel request that has not yet been consumed by a spin.
+  /**
+   * Set by cancel() and cleared when the spin it cancels (or, if none is in
+   * progress, the next spin) returns. The spin loops react to this flag only.
+   */
+  std::atomic_bool cancel_requested_;
+
   /// Guard condition for signaling the rmw layer to wake up for special events.
   std::shared_ptr<rclcpp::GuardCondition> interrupt_guard_condition_;
 

rclcpp/include/rclcpp/executors/events_cbg_executor/events_cbg_executor.hpp

@@ -205,8 +205,13 @@ class EventsCBGExecutor : public rclcpp::Executor
     if (spinning.exchange(true)) {
       throw std::runtime_error("spin_until_future_complete() called while already spinning");
     }
-    RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
-    while (rclcpp::ok(this->context_) && spinning.load()) {
+    RCPPUTILS_SCOPE_EXIT(
+      this->spinning.store(false);
+      this->cancel_requested_.store(false); );
+    if (cancel_requested_.load()) {
+      return FutureReturnCode::INTERRUPTED;
+    }
+    while (rclcpp::ok(this->context_) && !cancel_requested_.load()) {
       // Do one item of work.
       spin_once_internal(timeout_left);
 
@@ -315,7 +320,12 @@ class EventsCBGExecutor : public rclcpp::Executor
 
   std::atomic_bool needs_callback_group_resync = false;
 
-  /// Spinning state, used to prevent multi threaded calls to spin and to cancel blocking spins.
+  /// Spinning state, used to prevent multi threaded calls to spin.
+  /**
+   * This flag is only set and cleared by the spin functions themselves, so it
+   * stays true until a spin actually returns, even after cancel() was called.
+   * Cancellation is signaled via cancel_requested_ (inherited from Executor).
+   */
   std::atomic_bool spinning;
 
   /// set if we are shutting down.

rclcpp/src/rclcpp/executor.cpp

@@ -53,6 +53,7 @@ class rclcpp::ExecutorImplementation {};
 
 Executor::Executor(const std::shared_ptr<rclcpp::Context> & context)
 : spinning(false),
+  cancel_requested_(false),
   context_(context),
   entities_need_rebuild_(true),
   collector_(nullptr),
@@ -62,6 +63,7 @@ Executor::Executor(const std::shared_ptr<rclcpp::Context> & context)
 
 Executor::Executor(const rclcpp::ExecutorOptions & options)
 : spinning(false),
+  cancel_requested_(false),
   interrupt_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
   shutdown_guard_condition_(std::make_shared<rclcpp::GuardCondition>(options.context)),
   context_(options.context),
@@ -287,8 +289,14 @@ Executor::spin_until_future_complete_impl(
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_until_future_complete() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
-  while (rclcpp::ok(this->context_) && spinning.load()) {
+  RCPPUTILS_SCOPE_EXIT(
+    wait_result_.reset();
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return FutureReturnCode::INTERRUPTED;
+  }
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load()) {
     // Do one item of work.
     spin_once_impl(timeout_left);
 
@@ -378,7 +386,13 @@ Executor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_some() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
+  RCPPUTILS_SCOPE_EXIT(
+    wait_result_.reset();
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
 
   // clear the wait result and wait for work without blocking to collect the work
   // for the first time
@@ -396,7 +410,7 @@ Executor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
 
   // The logic of this while loop is as follows:
   //
-  // - while not shutdown, and spinning (not canceled), and not max duration reached...
+  // - while not shutdown, and not canceled, and not max duration reached...
   // - try to get an executable item to execute, and execute it if available
   // - otherwise, reset the wait result, and ...
   // - if there was no work available just after waiting, break the loop unconditionally
@@ -410,7 +424,7 @@ Executor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhaustive)
   // See also:
   //   https://github.com/ros2/rclcpp/issues/2508
   //   https://github.com/ros2/rclcpp/pull/2517
-  while (rclcpp::ok(context_) && spinning.load() && max_duration_not_elapsed()) {
+  while (rclcpp::ok(context_) && !cancel_requested_.load() && max_duration_not_elapsed()) {
     AnyExecutable any_exec;
     if (get_next_ready_executable(any_exec)) {
       execute_any_executable(any_exec);
@@ -460,14 +474,24 @@ Executor::spin_once(std::chrono::nanoseconds timeout)
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin_once() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
+  RCPPUTILS_SCOPE_EXIT(
+    wait_result_.reset();
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
   spin_once_impl(timeout);
 }
 
 void
 Executor::cancel()
 {
-  spinning.store(false);
+  // Only request the cancellation; the spinning flag is owned by the spin
+  // functions and is cleared when they actually return.  This keeps
+  // is_spinning() true until the executor has really stopped, and a cancel
+  // issued before a spin is "held" until the next spin consumes it.
+  cancel_requested_.store(true);
   try {
     interrupt_guard_condition_->trigger();
   } catch (const rclcpp::exceptions::RCLError & ex) {
@@ -479,7 +503,7 @@ Executor::cancel()
 void
 Executor::execute_any_executable(AnyExecutable & any_exec)
 {
-  if (!spinning.load()) {
+  if (cancel_requested_.load()) {
     return;
   }
 
@@ -920,7 +944,7 @@ Executor::get_next_executable(AnyExecutable & any_executable, std::chrono::nanos
   if (!success) {
     // Wait for subscriptions or timers to work on
     wait_for_work(timeout);
-    if (!spinning.load()) {
+    if (cancel_requested_.load()) {
       return false;
     }
     // Try again

rclcpp/src/rclcpp/executors/events_cbg_executor/events_cbg_executor.cpp

@@ -133,7 +133,7 @@ void EventsCBGExecutor::shutdown()
   bool was_spinning = spinning;
 
   // signal all processing threads to shut down
-  spinning = false;
+  cancel_requested_ = true;
 
   if(was_spinning) {
     scheduler->release_all_worker_threads();
@@ -342,7 +342,7 @@ EventsCBGExecutor::run(size_t this_thread_number, bool block_initially)
 {
   (void) this_thread_number;
 
-  while (rclcpp::ok(this->context_) && spinning.load() ) {
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load() ) {
     if(block_initially) {
       block_initially = false;
       scheduler->block_worker_thread();
@@ -373,7 +373,7 @@ EventsCBGExecutor::run(
 {
   (void) this_thread_number;
 
-  while (rclcpp::ok(this->context_) && spinning.load() ) {
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load() ) {
     sync_callback_groups();
 
     auto ready_entity = scheduler->get_next_ready_entity();
@@ -395,7 +395,7 @@ EventsCBGExecutor::run(
 
 void EventsCBGExecutor::spin_once_internal(std::chrono::nanoseconds timeout)
 {
-  if (!rclcpp::ok(this->context_) || !spinning.load() ) {
+  if (!rclcpp::ok(this->context_) || cancel_requested_.load() ) {
     return;
   }
 
@@ -430,7 +430,12 @@ EventsCBGExecutor::spin_once(std::chrono::nanoseconds timeout)
   if (spinning.exchange(true) ) {
     throw std::runtime_error("spin_once() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
 
   spin_once_internal(timeout);
 }
@@ -458,15 +463,20 @@ bool EventsCBGExecutor::collect_and_execute_ready_events(
   if (spinning.exchange(true) ) {
     throw std::runtime_error("collect_and_execute_ready_events() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return false;
+  }
 
   const auto start = std::chrono::steady_clock::now();
   const auto end_time = start + max_duration;
   auto cur_time = start;
 
   bool had_work = false;
 
-  while (rclcpp::ok(this->context_) && spinning && cur_time <= end_time) {
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load() && cur_time <= end_time) {
     sync_callback_groups();
 
     if (!execute_previous_ready_executables_until(end_time) ) {
@@ -491,7 +501,12 @@ EventsCBGExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
   std::vector<std::thread> threads;
   size_t thread_id = 0;
   for ( ; thread_id < number_of_threads_ - 1; ++thread_id) {
@@ -513,7 +528,12 @@ void EventsCBGExecutor::spin(
   if (spinning.exchange(true) ) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
   std::vector<std::thread> threads;
   size_t thread_id = 0;
   for ( ; thread_id < number_of_threads_ - 1; ++thread_id) {
@@ -562,11 +582,13 @@ EventsCBGExecutor::add_callback_group(
 void
 EventsCBGExecutor::cancel()
 {
-  bool was_spinning = spinning;
+  // Only request the cancellation; the spinning flag is owned by the spin
+  // functions and is cleared when they actually return.  This keeps
+  // is_spinning() true until the executor has really stopped, and a cancel
+  // issued before a spin is "held" until the next spin consumes it.
+  cancel_requested_.store(true);
 
-  spinning.store(false);
-
-  if(was_spinning) {
+  if(spinning) {
     scheduler->release_all_worker_threads();
   }
 

rclcpp/src/rclcpp/executors/multi_threaded_executor.cpp

@@ -55,7 +55,13 @@ MultiThreadedExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
+  RCPPUTILS_SCOPE_EXIT(
+    wait_result_.reset();
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
   std::vector<std::thread> threads;
   size_t thread_id = 0;
   {
@@ -81,11 +87,11 @@ MultiThreadedExecutor::get_number_of_threads()
 void
 MultiThreadedExecutor::run([[maybe_unused]] size_t this_thread_number)
 {
-  while (rclcpp::ok(this->context_) && spinning.load()) {
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load()) {
     rclcpp::AnyExecutable any_exec;
     {
       std::lock_guard wait_lock{wait_mutex_};
-      if (!rclcpp::ok(this->context_) || !spinning.load()) {
+      if (!rclcpp::ok(this->context_) || cancel_requested_.load()) {
         return;
       }
       if (!get_next_executable(any_exec, next_exec_timeout_)) {

rclcpp/src/rclcpp/executors/single_threaded_executor.cpp

@@ -30,13 +30,19 @@ SingleThreadedExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(wait_result_.reset();this->spinning.store(false););
+  RCPPUTILS_SCOPE_EXIT(
+    wait_result_.reset();
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
 
   // Clear any previous result and rebuild the waitset
   this->wait_result_.reset();
   this->entities_need_rebuild_ = true;
 
-  while (rclcpp::ok(this->context_) && spinning.load()) {
+  while (rclcpp::ok(this->context_) && !cancel_requested_.load()) {
     rclcpp::AnyExecutable any_executable;
     if (get_next_executable(any_executable)) {
       execute_any_executable(any_executable);

rclcpp/src/rclcpp/experimental/executors/events_executor/events_executor.cpp

@@ -103,7 +103,7 @@ EventsExecutor::setup_notify_waitable()
 
 EventsExecutor::~EventsExecutor()
 {
-  spinning.store(false);
+  cancel_requested_.store(true);
   notify_waitable_->clear_on_ready_callback();
   this->refresh_current_collection({});
 }
@@ -114,12 +114,17 @@ EventsExecutor::spin()
   if (spinning.exchange(true)) {
     throw std::runtime_error("spin() called while already spinning");
   }
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
 
   timers_manager_->start();
   RCPPUTILS_SCOPE_EXIT(timers_manager_->stop(); );
 
-  while (rclcpp::ok(context_) && spinning.load()) {
+  while (rclcpp::ok(context_) && !cancel_requested_.load()) {
     // Wait until we get an event
     ExecutorEvent event;
     bool has_event = events_queue_->dequeue(event);
@@ -151,7 +156,12 @@ EventsExecutor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhau
     throw std::runtime_error("spin_some() called while already spinning");
   }
 
-  RCPPUTILS_SCOPE_EXIT(this->spinning.store(false); );
+  RCPPUTILS_SCOPE_EXIT(
+    this->spinning.store(false);
+    this->cancel_requested_.store(false););
+  if (cancel_requested_.load()) {
+    return;
+  }
 
   auto start = std::chrono::steady_clock::now();
 
@@ -181,7 +191,7 @@ EventsExecutor::spin_some_impl(std::chrono::nanoseconds max_duration, bool exhau
   const size_t ready_timers_at_start = timers_manager_->get_number_ready_timers();
   size_t executed_timers = 0;
 
-  while (rclcpp::ok(context_) && spinning.load() && max_duration_not_elapsed()) {
+  while (rclcpp::ok(context_) && !cancel_requested_.load() && max_duration_not_elapsed()) {
     // Execute first ready event from queue if exists
     if (exhaustive || (executed_events < ready_events_at_start)) {
       bool has_event = !events_queue_->empty();