retry_scheduler.h

Go to the documentation of this file.

 
#pragma once
 
// system includes
#include <algorithm>
#include <condition_variable>
#include <functional>
#include <memory>
#include <thread>
 
// local includes
#include "logging.h"
 
namespace display_device {
  class SchedulerStopToken final {
  public:
    explicit SchedulerStopToken(std::function<void()> cleanup);
 
    SchedulerStopToken(const SchedulerStopToken &) = delete;
 
    SchedulerStopToken &operator=(const SchedulerStopToken &) = delete;
 
    ~SchedulerStopToken();
 
    void requestStop();
 
    [[nodiscard]] bool stopRequested() const;
 
  private:
    bool m_stop_requested {false};
    std::function<void()> m_cleanup;
  };
 
  namespace detail {
    template<class FunctionT>
    concept OptionalFunction = requires(FunctionT exec_fn) {
      static_cast<bool>(exec_fn);
    };
 
    template<class T, bool AddConst>
    struct AutoConst;
 
    template<class T>
    struct AutoConst<T, false> {
      using type = T;
    };
 
    template<class T>
    struct AutoConst<T, true> {
      using type = std::add_const_t<T>;
    };
 
    template<class T, bool AddConst>
    using auto_const_t = typename AutoConst<T, AddConst>::type;
 
    template<class T, class FunctionT>
    concept ExecuteWithoutStopToken = requires(FunctionT exec_fn, T &value) { exec_fn(value); };
 
    template<class T, class FunctionT>
    concept ExecuteWithStopToken = requires(FunctionT exec_fn, T &value, SchedulerStopToken &token) {
      exec_fn(value, token);
    };
 
    template<class T, class FunctionT>
    concept ExecuteCallbackLike = ExecuteWithoutStopToken<T, FunctionT> || ExecuteWithStopToken<T, FunctionT>;
  }  // namespace detail
 
  struct SchedulerOptions {
    enum class Execution {
      Immediate,  
      ImmediateWithSleep,  
      ScheduledOnly  
    };
 
    std::vector<std::chrono::milliseconds> m_sleep_durations;  
    Execution m_execution {Execution::Immediate};  
  };
 
  template<class T>
  class RetryScheduler final {
  public:
    explicit RetryScheduler(std::unique_ptr<T> iface):
        m_iface {iface ? std::move(iface) : throw std::logic_error {"Nullptr interface provided in RetryScheduler!"}},
        m_thread {[this]() {
          std::unique_lock lock {m_mutex};
          while (m_keep_alive) {
            m_syncing_thread = false;
            if (auto duration {takeNextDuration(m_sleep_durations)}; duration > std::chrono::milliseconds::zero()) {
              // We're going to sleep until manually woken up or the time elapses.
              m_sleep_cv.wait_for(lock, duration, [this]() {
                return m_syncing_thread;
              });
            } else {
              // We're going to sleep until manually woken up.
              m_sleep_cv.wait(lock, [this]() {
                return m_syncing_thread;
              });
            }
 
            if (m_syncing_thread) {
              // Thread was waken up to sync sleep time or to be stopped.
              continue;
            }
 
            try {
              SchedulerStopToken scheduler_stop_token {[&]() {
                clearThreadLoopUnlocked();
              }};
              m_retry_function(*m_iface, scheduler_stop_token);
              continue;
            } catch (const std::exception &error) {
              DD_LOG(error) << "Exception thrown in the RetryScheduler thread. Stopping scheduler. Error:\n"
                            << error.what();
            }
 
            clearThreadLoopUnlocked();
          }
        }} {
    }
 
    ~RetryScheduler() {
      {
        std::lock_guard lock {m_mutex};
        m_keep_alive = false;
        syncThreadUnlocked();
      }
 
      m_thread.join();
    }
 
    void schedule(std::function<void(T &, SchedulerStopToken &stop_token)> exec_fn, const SchedulerOptions &options) {
      if (!exec_fn) {
        throw std::logic_error {"Empty callback function provided in RetryScheduler::schedule!"};
      }
 
      if (options.m_sleep_durations.empty()) {
        throw std::logic_error {"At least 1 sleep duration must be specified in RetryScheduler::schedule!"};
      }
 
      if (std::ranges::any_of(options.m_sleep_durations, [&](const auto &duration) {
            return duration == std::chrono::milliseconds::zero();
          })) {
        throw std::logic_error {"All of the durations specified in RetryScheduler::schedule must be larger than a 0!"};
      }
 
      std::lock_guard lock {m_mutex};
      SchedulerStopToken stop_token {[&]() {
        stopUnlocked();
      }};
 
      // We are catching the exception here instead of propagating to have
      // similar try...catch login as in the scheduler thread.
      try {
        auto sleep_durations = options.m_sleep_durations;
        if (options.m_execution != SchedulerOptions::Execution::ScheduledOnly) {
          if (options.m_execution == SchedulerOptions::Execution::ImmediateWithSleep) {
            std::this_thread::sleep_for(takeNextDuration(sleep_durations));
          }
 
          exec_fn(*m_iface, stop_token);
        }
 
        if (!stop_token.stopRequested()) {
          m_retry_function = std::move(exec_fn);
          m_sleep_durations = std::move(sleep_durations);
          syncThreadUnlocked();
        }
      } catch (const std::exception &error) {
        stop_token.requestStop();
        DD_LOG(error) << "Exception thrown in the RetryScheduler::schedule. Stopping scheduler. Error:\n"
                      << error.what();
      }
    }
 
    template<class FunctionT>
    auto execute(FunctionT &&exec_fn) {
      return executeImpl(*this, std::forward<FunctionT>(exec_fn));
    }
 
    template<class FunctionT>
    auto execute(FunctionT &&exec_fn) const {
      return executeImpl(*this, std::forward<FunctionT>(exec_fn));
    }
 
    [[nodiscard]] bool isScheduled() const {
      return static_cast<bool>(m_retry_function);
    }
 
    void stop() {
      std::lock_guard lock {m_mutex};
      stopUnlocked();
    }
 
  private:
    static std::chrono::milliseconds takeNextDuration(std::vector<std::chrono::milliseconds> &durations) {
      if (durations.size() > 1) {
        const auto front_it {std::begin(durations)};
        const auto front_value {*front_it};
        durations.erase(front_it);
        return front_value;
      }
 
      return durations.empty() ? std::chrono::milliseconds::zero() : durations.back();
    }
 
    static auto executeImpl(auto &self, auto &&exec_fn)
      requires detail::ExecuteCallbackLike<T, decltype(exec_fn)>
    {
      using FunctionT = decltype(exec_fn);
      constexpr bool IsConst = std::is_const_v<std::remove_reference_t<decltype(self)>>;
 
      if constexpr (detail::OptionalFunction<FunctionT>) {
        if (!exec_fn) {
          throw std::logic_error {"Empty callback function provided in RetryScheduler::execute!"};
        }
      }
 
      std::lock_guard lock {self.m_mutex};
      detail::auto_const_t<std::decay_t<T>, IsConst> &iface_ref {*self.m_iface};
      if constexpr (detail::ExecuteWithStopToken<T, FunctionT>) {
        detail::auto_const_t<SchedulerStopToken, IsConst> stop_token {[&self]() {
          if constexpr (!IsConst) {
            self.stopUnlocked();
          }
        }};
        return std::forward<FunctionT>(exec_fn)(iface_ref, stop_token);
      } else {
        return std::forward<FunctionT>(exec_fn)(iface_ref);
      }
    }
 
    void clearThreadLoopUnlocked() {
      m_sleep_durations = {};
      m_retry_function = nullptr;
    }
 
    void syncThreadUnlocked() {
      m_syncing_thread = true;
      m_sleep_cv.notify_one();
    }
 
    void stopUnlocked() {
      if (isScheduled()) {
        clearThreadLoopUnlocked();
        syncThreadUnlocked();
      }
    }
 
    std::unique_ptr<T> m_iface; 
    std::vector<std::chrono::milliseconds> m_sleep_durations; 
    std::function<void(T &, SchedulerStopToken &)> m_retry_function {nullptr}; 
    mutable std::mutex m_mutex {}; 
    std::condition_variable m_sleep_cv {}; 
    bool m_syncing_thread {false}; 
    bool m_keep_alive {true}; 
    // Always the last in the list so that all the members are already initialized!
    std::thread m_thread; 
  };
}  // namespace display_device