Address review: remove mutable, use remove_if, fix style

- Remove mutable from entity vectors, make collect_all_ptrs non-const
- Replace collect_and_compact helper with std::remove_if (erase-remove idiom)
- Remove leaked test_time_source_deadlock CMake entry from another branch
- Fix uncrustify formatting in test file

Signed-off-by: Pavel Guzenfeld <pavelguzenfeld@gmail.com>

Author: PavelGuzenfeld

rclcpp/include/rclcpp/callback_group.hpp

@@ -183,7 +183,7 @@ class CallbackGroup
     const std::function<void(const rclcpp::ServiceBase::SharedPtr &)> & service_func,
     const std::function<void(const rclcpp::ClientBase::SharedPtr &)> & client_func,
     const std::function<void(const rclcpp::TimerBase::SharedPtr &)> & timer_func,
-    const std::function<void(const rclcpp::Waitable::SharedPtr &)> & waitable_func) const;
+    const std::function<void(const rclcpp::Waitable::SharedPtr &)> & waitable_func);
 
   /// Return a reference to the 'associated with executor' atomic boolean.
   /**
@@ -258,11 +258,11 @@ class CallbackGroup
   // Mutex to protect the subsequent vectors of pointers.
   mutable std::mutex mutex_;
   std::atomic_bool associated_with_executor_;
-  mutable std::vector<rclcpp::SubscriptionBase::WeakPtr> subscription_ptrs_;
-  mutable std::vector<rclcpp::TimerBase::WeakPtr> timer_ptrs_;
-  mutable std::vector<rclcpp::ServiceBase::WeakPtr> service_ptrs_;
-  mutable std::vector<rclcpp::ClientBase::WeakPtr> client_ptrs_;
-  mutable std::vector<rclcpp::Waitable::WeakPtr> waitable_ptrs_;
+  std::vector<rclcpp::SubscriptionBase::WeakPtr> subscription_ptrs_;
+  std::vector<rclcpp::TimerBase::WeakPtr> timer_ptrs_;
+  std::vector<rclcpp::ServiceBase::WeakPtr> service_ptrs_;
+  std::vector<rclcpp::ClientBase::WeakPtr> client_ptrs_;
+  std::vector<rclcpp::Waitable::WeakPtr> waitable_ptrs_;
   std::atomic_bool can_be_taken_from_;
   const bool automatically_add_to_executor_with_node_;
   // defer the creation of the guard condition

rclcpp/src/rclcpp/callback_group.cpp

@@ -68,44 +68,74 @@ CallbackGroup::size() const
     waitable_ptrs_.size();
 }
 
-namespace
-{
-/// Iterate a vector of weak pointers, call func for each live entry,
-/// and compact expired entries out of the vector in a single pass.
-template<typename T, typename Func>
-void collect_and_compact(
-  std::vector<typename T::WeakPtr> & ptrs,
-  const Func & func)
-{
-  size_t write_idx = 0;
-  for (size_t read_idx = 0; read_idx < ptrs.size(); ++read_idx) {
-    auto ref_ptr = ptrs[read_idx].lock();
-    if (ref_ptr) {
-      func(ref_ptr);
-      if (write_idx != read_idx) {
-        ptrs[write_idx] = std::move(ptrs[read_idx]);
-      }
-      ++write_idx;
-    }
-  }
-  ptrs.resize(write_idx);
-}
-}  // namespace
-
 void CallbackGroup::collect_all_ptrs(
   const std::function<void(const rclcpp::SubscriptionBase::SharedPtr &)> & sub_func,
   const std::function<void(const rclcpp::ServiceBase::SharedPtr &)> & service_func,
   const std::function<void(const rclcpp::ClientBase::SharedPtr &)> & client_func,
   const std::function<void(const rclcpp::TimerBase::SharedPtr &)> & timer_func,
-  const std::function<void(const rclcpp::Waitable::SharedPtr &)> & waitable_func) const
+  const std::function<void(const rclcpp::Waitable::SharedPtr &)> & waitable_func)
 {
   std::lock_guard<std::mutex> lock(mutex_);
 
-  collect_and_compact<rclcpp::SubscriptionBase>(subscription_ptrs_, sub_func);
-  collect_and_compact<rclcpp::ServiceBase>(service_ptrs_, service_func);
-  collect_and_compact<rclcpp::ClientBase>(client_ptrs_, client_func);
-  collect_and_compact<rclcpp::TimerBase>(timer_ptrs_, timer_func);
-  collect_and_compact<rclcpp::Waitable>(waitable_ptrs_, waitable_func);
+  for (const auto & weak_ptr : subscription_ptrs_) {
+    auto ref_ptr = weak_ptr.lock();
+    if (ref_ptr) {
+      sub_func(ref_ptr);
+    }
+  }
+  subscription_ptrs_.erase(
+    std::remove_if(
+      subscription_ptrs_.begin(), subscription_ptrs_.end(),
+      [](const auto & w) {return w.expired();}),
+    subscription_ptrs_.end());
+
+  for (const auto & weak_ptr : service_ptrs_) {
+    auto ref_ptr = weak_ptr.lock();
+    if (ref_ptr) {
+      service_func(ref_ptr);
+    }
+  }
+  service_ptrs_.erase(
+    std::remove_if(
+      service_ptrs_.begin(), service_ptrs_.end(),
+      [](const auto & w) {return w.expired();}),
+    service_ptrs_.end());
+
+  for (const auto & weak_ptr : client_ptrs_) {
+    auto ref_ptr = weak_ptr.lock();
+    if (ref_ptr) {
+      client_func(ref_ptr);
+    }
+  }
+  client_ptrs_.erase(
+    std::remove_if(
+      client_ptrs_.begin(), client_ptrs_.end(),
+      [](const auto & w) {return w.expired();}),
+    client_ptrs_.end());
+
+  for (const auto & weak_ptr : timer_ptrs_) {
+    auto ref_ptr = weak_ptr.lock();
+    if (ref_ptr) {
+      timer_func(ref_ptr);
+    }
+  }
+  timer_ptrs_.erase(
+    std::remove_if(
+      timer_ptrs_.begin(), timer_ptrs_.end(),
+      [](const auto & w) {return w.expired();}),
+    timer_ptrs_.end());
+
+  for (const auto & weak_ptr : waitable_ptrs_) {
+    auto ref_ptr = weak_ptr.lock();
+    if (ref_ptr) {
+      waitable_func(ref_ptr);
+    }
+  }
+  waitable_ptrs_.erase(
+    std::remove_if(
+      waitable_ptrs_.begin(), waitable_ptrs_.end(),
+      [](const auto & w) {return w.expired();}),
+    waitable_ptrs_.end());
 }
 
 std::atomic_bool &

rclcpp/test/rclcpp/CMakeLists.txt

@@ -442,12 +442,6 @@ ament_add_ros_isolated_gtest(test_time_source test_time_source.cpp
 if(TARGET test_time_source)
   target_link_libraries(test_time_source ${PROJECT_NAME} rcl::rcl)
 endif()
-ament_add_ros_isolated_gtest(test_time_source_deadlock test_time_source_deadlock.cpp
-  TIMEOUT 60)
-if(TARGET test_time_source_deadlock)
-  target_link_libraries(test_time_source_deadlock ${PROJECT_NAME})
-endif()
-
 ament_add_ros_isolated_gtest(test_utilities test_utilities.cpp
   APPEND_LIBRARY_DIRS "${append_library_dirs}")
 ament_add_test_label(test_utilities mimick)

rclcpp/test/rclcpp/test_callback_group_entity_cleanup.cpp

@@ -182,7 +182,7 @@ TEST_F(TestCallbackGroupEntityCleanup, add_many_timers_is_not_quadratic)
   // With O(N²) the old code took ~429ms for 10K timers.
   // With O(N) we expect < 100ms for 5K timers even on slow CI.
   // Use a generous threshold to avoid flakiness.
-  EXPECT_LT(ms, 5000) << "Adding " << N << " timers took " << ms
+  EXPECT_LT(ms, 5000)  << "Adding " << N << " timers took " << ms
                        << "ms — possible quadratic regression";
 }
 
@@ -215,7 +215,7 @@ TEST_F(TestCallbackGroupEntityCleanup, interleaved_add_remove_cycles)
     [](const rclcpp::SubscriptionBase::SharedPtr &) {},
     [](const rclcpp::ServiceBase::SharedPtr &) {},
     [](const rclcpp::ClientBase::SharedPtr &) {},
-    [&live](const rclcpp::TimerBase::SharedPtr &) { ++live; },
+    [&live](const rclcpp::TimerBase::SharedPtr &) {++live;},
     [](const rclcpp::Waitable::SharedPtr &) {});
 
   EXPECT_EQ(live, timers.size());