RecurrentSharedFuture.h

浏览该文件的文档.

// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
// de Barcelona (UAB).
//
// This work is licensed under the terms of the MIT license.
// For a copy, see <https://opensource.org/licenses/MIT>.
 
#pragma once
 
#include "carla/Exception.h"
#include "carla/Time.h"
 
#include <boost/optional.hpp>
#ifdef _MSC_VER
#pragma warning(push)
#pragma warning(disable:4583)
#pragma warning(disable:4582)
#include <boost/variant2/variant.hpp>
#pragma warning(pop)
#else
#include <boost/variant2/variant.hpp>
#endif
 
#include <condition_variable>
#include <exception>
#include <map>
#include <mutex>
 
namespace carla {
 
namespace detail {
 
  class SharedException;
 
} // namespace detail
 
  // ===========================================================================
  // -- RecurrentSharedFuture --------------------------------------------------
  // ===========================================================================
 
  /// This class is meant to be used similar to a shared future, but the value
  /// can be set any number of times.
  template <typename T>
  class RecurrentSharedFuture {
  public:
 
    using SharedException = detail::SharedException;
 
    /// Wait until the next value is set. Any number of threads can be waiting
    /// simultaneously.
    ///
    /// @return empty optional if the timeout is met.
    boost::optional<T> WaitFor(time_duration timeout);
 
    /// Set the value and notify all waiting threads.
    template <typename T2>
    void SetValue(const T2 &value);
 
    /// Set a exception, this exception will be thrown on all the threads
    /// waiting.
    ///
    /// @note The @a exception will be stored on a SharedException and thrown
    /// as such.
    template <typename ExceptionT>
    void SetException(ExceptionT &&exception);
 
  private:
 
    std::mutex _mutex;
 
    std::condition_variable _cv;
 
    struct mapped_type {
      bool should_wait;
      boost::variant2::variant<SharedException, T> value;
    };
 
    std::map<const char *, mapped_type> _map;
  };
 
  // ===========================================================================
  // -- RecurrentSharedFuture implementation -----------------------------------
  // ===========================================================================
 
namespace detail {
 
  static thread_local const char thread_tag{};
 
  class SharedException : public std::exception {
  public:
 
    SharedException()
      : _exception(std::make_shared<std::runtime_error>("uninitialized SharedException")) {}
 
    SharedException(std::shared_ptr<std::exception> e)
      : _exception(std::move(e)) {}
 
    const char *what() const noexcept override {
      return _exception->what();
    }
 
    std::shared_ptr<std::exception> GetException() const {
      return _exception;
    }
 
  private:
 
    std::shared_ptr<std::exception> _exception;
  };
 
} // namespace detail
 
  template <typename T>
  boost::optional<T> RecurrentSharedFuture<T>::WaitFor(time_duration timeout) {
    std::unique_lock<std::mutex> lock(_mutex);
    auto &r = _map[&detail::thread_tag];
    r.should_wait = true;
    if (!_cv.wait_for(lock, timeout.to_chrono(), [&]() { return !r.should_wait; })) {
      return {};
    }
    if (r.value.index() == 0) {
      throw_exception(boost::variant2::get<SharedException>(r.value));
    }
    return boost::variant2::get<T>(std::move(r.value));
  }
 
  template <typename T>
  template <typename T2>
  void RecurrentSharedFuture<T>::SetValue(const T2 &value) {
    std::lock_guard<std::mutex> lock(_mutex);
    for (auto &pair : _map) {
      pair.second.should_wait = false;
      pair.second.value = value;
    }
    _cv.notify_all();
  }
 
  template <typename T>
  template <typename ExceptionT>
  void RecurrentSharedFuture<T>::SetException(ExceptionT &&e) {
    SetValue(SharedException(std::make_shared<ExceptionT>(std::forward<ExceptionT>(e))));
  }
 
} // namespace carla