Re-add Clock::sleep_until (ros2#1814)

* Revert "Revert "Add Clock::sleep_until method (ros2#1748)" (ros2#1793)"

This reverts commit d04319a.

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Context, Shutdown Callback, Condition Var per call

The `Clock` doesn't have enough information to know which Context should
wake it on shutdown, so this adds a Context as an argument to
sleep_until().

Since the context is per call, the shutdown callback is also registered
per call and cannot be stored on impl_.

The condition_variable is also unique per call to reduce spurious
wakeups when multiple threads sleep on the same clock.

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Throw if until has wrong clock type

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* unnecessary newline

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Shorten line

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Fix time jump thresholds and add ROS time test

Use -1 and 1 thresholds because 0 and 0 is supposed to disable the
callbacks

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Shorten line

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* rclcpp::ok() -> context->is_valid()

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* No pre-jump handler instead of noop handler

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* If ros_time_is_active errors, let it throw

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Get time source change from callback to avoid race if ROS time toggled quickly

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Fix threshold and no  pre-jump callback

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Use RCUTILS_MS_TO_NS macro

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Explicit cast for duration to system time

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Throws at the end of docblock

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

* Add tests for invalid and non-global contexts

Signed-off-by: Shane Loretz <sloretz@osrfoundation.org>

rclcpp/include/rclcpp/clock.hpp

@@ -19,6 +19,7 @@
 #include <memory>
 #include <mutex>
 
+#include "rclcpp/contexts/default_context.hpp"
 #include "rclcpp/macros.hpp"
 #include "rclcpp/time.hpp"
 #include "rclcpp/visibility_control.hpp"
@@ -78,6 +79,30 @@ class Clock
   Time
   now();
 
+  /**
+   * Sleep until a specified Time, according to clock type.
+   *
+   * Notes for RCL_ROS_TIME clock type:
+   *   - Can sleep forever if ros time is active and received clock never reaches `until`
+   *   - If ROS time enabled state changes during the sleep, this method will immediately return
+   *     false. There is not a consistent choice of sleeping time when the time source changes,
+   *     so this is up to the caller to call again if needed.
+   *
+   * \param until absolute time according to current clock type to sleep until.
+   * \param context the rclcpp context the clock should use to check that ROS is still initialized.
+   * \return true immediately if `until` is in the past
+   * \return true when the time `until` is reached
+   * \return false if time cannot be reached reliably, for example from shutdown or a change
+   *    of time source.
+   * \throws std::runtime_error if the context is invalid
+   * \throws std::runtime_error if `until` has a different clock type from this clock
+   */
+  RCLCPP_PUBLIC
+  bool
+  sleep_until(
+    Time until,
+    Context::SharedPtr context = contexts::get_global_default_context());
+
   /**
    * Returns the clock of the type `RCL_ROS_TIME` is active.
    *

rclcpp/src/rclcpp/clock.cpp

@@ -14,11 +14,14 @@
 
 #include "rclcpp/clock.hpp"
 
+#include <condition_variable>
 #include <memory>
 #include <thread>
 
 #include "rclcpp/exceptions.hpp"
+#include "rclcpp/utilities.hpp"
 
+#include "rcpputils/scope_exit.hpp"
 #include "rcutils/logging_macros.h"
 
 namespace rclcpp
@@ -76,6 +79,99 @@ Clock::now()
   return now;
 }
 
+bool
+Clock::sleep_until(Time until, Context::SharedPtr context)
+{
+  if (!context || !context->is_valid()) {
+    throw std::runtime_error("context cannot be slept with because it's invalid");
+  }
+  const auto this_clock_type = get_clock_type();
+  if (until.get_clock_type() != this_clock_type) {
+    throw std::runtime_error("until's clock type does not match this clock's type");
+  }
+  bool time_source_changed = false;
+
+  std::condition_variable cv;
+
+  // Wake this thread if the context is shutdown
+  rclcpp::OnShutdownCallbackHandle shutdown_cb_handle = context->add_on_shutdown_callback(
+    [&cv]() {
+      cv.notify_one();
+    });
+  // No longer need the shutdown callback when this function exits
+  auto callback_remover = rcpputils::scope_exit(
+    [context, &shutdown_cb_handle]() {
+      context->remove_on_shutdown_callback(shutdown_cb_handle);
+    });
+
+  if (this_clock_type == RCL_STEADY_TIME) {
+    auto steady_time = std::chrono::steady_clock::time_point(
+      std::chrono::nanoseconds(until.nanoseconds()));
+
+    // loop over spurious wakeups but notice shutdown
+    std::unique_lock lock(impl_->clock_mutex_);
+    while (now() < until && context->is_valid()) {
+      cv.wait_until(lock, steady_time);
+    }
+  } else if (this_clock_type == RCL_SYSTEM_TIME) {
+    auto system_time = std::chrono::system_clock::time_point(
+      // Cast because system clock resolution is too big for nanoseconds on some systems
+      std::chrono::duration_cast<std::chrono::system_clock::duration>(
+        std::chrono::nanoseconds(until.nanoseconds())));
+
+    // loop over spurious wakeups but notice shutdown
+    std::unique_lock lock(impl_->clock_mutex_);
+    while (now() < until && context->is_valid()) {
+      cv.wait_until(lock, system_time);
+    }
+  } else if (this_clock_type == RCL_ROS_TIME) {
+    // Install jump handler for any amount of time change, for two purposes:
+    // - if ROS time is active, check if time reached on each new clock sample
+    // - Trigger via on_clock_change to detect if time source changes, to invalidate sleep
+    rcl_jump_threshold_t threshold;
+    threshold.on_clock_change = true;
+    // 0 is disable, so -1 and 1 are smallest possible time changes
+    threshold.min_backward.nanoseconds = -1;
+    threshold.min_forward.nanoseconds = 1;
+    auto clock_handler = create_jump_callback(
+      nullptr,
+      [&cv, &time_source_changed](const rcl_time_jump_t & jump) {
+        if (jump.clock_change != RCL_ROS_TIME_NO_CHANGE) {
+          time_source_changed = true;
+        }
+        cv.notify_one();
+      },
+      threshold);
+
+    if (!ros_time_is_active()) {
+      auto system_time = std::chrono::system_clock::time_point(
+        // Cast because system clock resolution is too big for nanoseconds on some systems
+        std::chrono::duration_cast<std::chrono::system_clock::duration>(
+          std::chrono::nanoseconds(until.nanoseconds())));
+
+      // loop over spurious wakeups but notice shutdown or time source change
+      std::unique_lock lock(impl_->clock_mutex_);
+      while (now() < until && context->is_valid() && !time_source_changed) {
+        cv.wait_until(lock, system_time);
+      }
+    } else {
+      // RCL_ROS_TIME with ros_time_is_active.
+      // Just wait without "until" because installed
+      // jump callbacks wake the cv on every new sample.
+      std::unique_lock lock(impl_->clock_mutex_);
+      while (now() < until && context->is_valid() && !time_source_changed) {
+        cv.wait(lock);
+      }
+    }
+  }
+
+  if (!context->is_valid() || time_source_changed) {
+    return false;
+  }
+
+  return now() >= until;
+}
+
 bool
 Clock::ros_time_is_active()
 {