realtime_publisher.hpp

// Copyright (c) 2008, Willow Garage, Inc.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//    * Redistributions of source code must retain the above copyright
//      notice, this list of conditions and the following disclaimer.
//
//    * Redistributions in binary form must reproduce the above copyright
//      notice, this list of conditions and the following disclaimer in the
//      documentation and/or other materials provided with the distribution.
//
//    * Neither the name of the Willow Garage, Inc. nor the names of its
//      contributors may be used to endorse or promote products derived from
//      this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.
 
/*
 * Publishing ROS messages is difficult, as the publish function is
 * not realtime safe.  This class provides the proper locking so that
 * you can call publish in realtime and a separate (non-realtime)
 * thread will ensure that the message gets published over ROS.
 *
 * Author: Stuart Glaser
 */
#ifndef REALTIME_TOOLS__REALTIME_PUBLISHER_HPP_
#define REALTIME_TOOLS__REALTIME_PUBLISHER_HPP_
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <utility>
 
#include "rclcpp/create_publisher.hpp"
#include "rclcpp/publisher.hpp"
 
namespace realtime_tools
{
template <class MessageT>
class RealtimePublisher
{
public:
  using PublisherType = rclcpp::Publisher<MessageT>;
  using PublisherSharedPtr = typename rclcpp::Publisher<MessageT>::SharedPtr;
 
  using PublishedType = typename rclcpp::TypeAdapter<MessageT>::custom_type;
  using ROSMessageType = typename rclcpp::TypeAdapter<MessageT>::ros_message_type;
 
  RCLCPP_SMART_PTR_DEFINITIONS(RealtimePublisher<MessageT>)
 
  MessageT msg_;
 
  template <typename NodeT>
  explicit RealtimePublisher(
    NodeT && node, const std::string & topic_name, const rclcpp::QoS & qos,
    const rclcpp::PublisherOptions & options = rclcpp::PublisherOptions())
  {
    initialize([&]() {
      return rclcpp::create_publisher<MessageT>(
        std::forward<NodeT>(node), topic_name, qos, options);
    });
  }
 
  explicit RealtimePublisher(PublisherSharedPtr publisher)
  {
    initialize([&]() { return publisher; });
  }
 
  [[deprecated(
    "Use constructor with rclcpp::Publisher<T>::SharedPtr instead - this class does not make sense "
    "without a real publisher")]]
  RealtimePublisher()
  : is_running_(false), keep_running_(false), turn_(State::LOOP_NOT_STARTED)
  {
  }
 
  ~RealtimePublisher()
  {
    RCLCPP_DEBUG(rclcpp::get_logger("realtime_tools"), "Waiting for publishing thread to stop....");
    stop();
    while (is_running()) {
      std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
    RCLCPP_DEBUG(
      rclcpp::get_logger("realtime_tools"), "Publishing thread stopped, joining thread....");
    if (thread_.joinable()) {
      thread_.join();
    }
  }
 
  void stop()
  {
    {
      std::unique_lock<std::mutex> lock(msg_mutex_);
      keep_running_ = false;
    }
    updated_cond_.notify_one();  // So the publishing loop can exit
  }
 
  bool trylock()
  {
    return turn_.load(std::memory_order_acquire) == State::REALTIME && msg_mutex_.try_lock();
  }
 
  bool can_publish() const
  {
    std::unique_lock<std::mutex> lock(msg_mutex_, std::try_to_lock);
    return can_publish(lock);
  }
 
  bool try_publish(const MessageT & msg)
  {
    std::unique_lock<std::mutex> lock(msg_mutex_, std::try_to_lock);
    if (can_publish(lock)) {
      {
        std::unique_lock<std::mutex> scoped_lock(std::move(lock));
        msg_ = msg;
        turn_.store(State::NON_REALTIME, std::memory_order_release);
      }
      updated_cond_.notify_one();  // Notify the publishing thread
      return true;
    }
    return false;
  }
 
  [[deprecated(
    "Use try_publish() method instead of this method. This method may be removed in future "
    "versions.")]]
  bool tryPublish(const MessageT & msg)
  {
    return try_publish(msg);
  }
 
  void unlockAndPublish()
  {
    turn_.store(State::NON_REALTIME, std::memory_order_release);
    unlock();
  }
 
  void lock() { msg_mutex_.lock(); }
 
  void unlock()
  {
    msg_mutex_.unlock();
    updated_cond_.notify_one();
  }
 
  std::thread & get_thread() { return thread_; }
 
  const std::thread & get_thread() const { return thread_; }
 
  const MessageT & get_msg() const { return msg_; }
 
  std::mutex & get_mutex() { return msg_mutex_; }
 
  const std::mutex & get_mutex() const { return msg_mutex_; }
 
private:
  template <typename PublisherCreator>
  void initialize(PublisherCreator && creator)
  {
    publisher_ = creator();
    is_running_ = false;
    keep_running_ = true;
    turn_ = State::LOOP_NOT_STARTED;
 
    thread_ = std::thread(&RealtimePublisher::publishingLoop, this);
 
    // Wait for the thread to be ready before proceeding
    // This is important to ensure that the thread is properly initialized and ready to handle
    // messages before any other operations are performed on the RealtimePublisher instance.
    while (!thread_.joinable() ||
           turn_.load(std::memory_order_acquire) == State::LOOP_NOT_STARTED) {
      std::this_thread::sleep_for(std::chrono::microseconds(100));
    }
  }
 
  bool can_publish(std::unique_lock<std::mutex> & lock) const
  {
    return turn_.load(std::memory_order_acquire) == State::REALTIME && lock.owns_lock();
  }
 
  // non-copyable
  RealtimePublisher(const RealtimePublisher &) = delete;
  RealtimePublisher & operator=(const RealtimePublisher &) = delete;
 
  bool is_running() const { return is_running_; }
 
  void publishingLoop()
  {
    is_running_ = true;
 
    while (keep_running_) {
      MessageT outgoing;
 
      {
        turn_.store(State::REALTIME, std::memory_order_release);
        // Locks msg_ and copies it to outgoing
        std::unique_lock<std::mutex> lock_(msg_mutex_);
        updated_cond_.wait(lock_, [&] { return turn_ == State::NON_REALTIME || !keep_running_; });
        outgoing = msg_;
      }
 
      // Sends the outgoing message
      if (keep_running_) {
        publisher_->publish(outgoing);
      }
    }
    is_running_ = false;
  }
 
  PublisherSharedPtr publisher_;
  std::atomic<bool> is_running_;
  std::atomic<bool> keep_running_;
 
  std::thread thread_;
 
  mutable std::mutex msg_mutex_;  // Protects msg_
  std::condition_variable updated_cond_;
 
  enum class State : int { REALTIME, NON_REALTIME, LOOP_NOT_STARTED };
  std::atomic<State> turn_;  // Who's turn is it to use msg_?
};
 
template <class MessageT>
using RealtimePublisherSharedPtr = std::shared_ptr<RealtimePublisher<MessageT>>;
 
}  // namespace realtime_tools
#endif  // REALTIME_TOOLS__REALTIME_PUBLISHER_HPP_