retry_scheduler.h

Go to the documentation of this file.

 
#pragma once
 
// system includes
#include <algorithm>
#include <condition_variable>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <stdexcept>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
 
// 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() noexcept;
 
    void requestStop();
 
    [[nodiscard]] bool stopRequested() const;
 
  private:
    bool m_stop_requested {false};
    std::function<void()> m_cleanup;
  };
 
  namespace detail {
    inline void logSchedulerException(const std::exception &exception, const char *message) {
      DD_LOG(error) << message << " Error:\n"
                    << exception.what();
    }
 
    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::invalid_argument {"Nullptr interface provided in RetryScheduler!"}},
        m_thread {[this]() {
          runThreadLoop();
        }} {
    }
 
    ~RetryScheduler() {
      {
        std::lock_guard lock {m_mutex};
        m_keep_alive = false;
        syncThreadUnlocked();
      }
 
      if (m_thread.joinable()) {
        m_thread.join();
      }
    }
 
    void schedule(std::function<void(T &, SchedulerStopToken &stop_token)> exec_fn, const SchedulerOptions &options) {
      if (!exec_fn) {
        throw std::invalid_argument {"Empty callback function provided in RetryScheduler::schedule!"};
      }
 
      if (options.m_sleep_durations.empty()) {
        throw std::invalid_argument {"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::invalid_argument {"All of the durations specified in RetryScheduler::schedule must be larger than a 0!"};
      }
 
      std::lock_guard lock {m_mutex};
      SchedulerStopToken stop_token {[this]() {
        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) {  // NOSONAR(cpp:S1181): Scheduler callback boundary must catch standard callback failures.
        stop_token.requestStop();
        detail::logSchedulerException(error, "Exception thrown in the RetryScheduler::schedule. Stopping scheduler.");
      }
    }
 
    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();
    }
 
    template<class SelfT, class FunctionT>
      requires detail::ExecuteCallbackLike<T, FunctionT>
    static auto executeImpl(SelfT &self, FunctionT &&exec_fn) {
      constexpr bool IsConst = std::is_const_v<std::remove_reference_t<SelfT>>;
 
      if constexpr (detail::OptionalFunction<FunctionT>) {
        if (!exec_fn) {
          throw std::invalid_argument {"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 {makeStopCallback(self)};
        return std::forward<FunctionT>(exec_fn)(iface_ref, stop_token);
      } else {
        return std::forward<FunctionT>(exec_fn)(iface_ref);
      }
    }
 
    template<class SelfT>
    static auto makeStopCallback(SelfT &self) {
      constexpr bool IsConst = std::is_const_v<std::remove_reference_t<SelfT>>;
      if constexpr (IsConst) {
        return []() {
          // Const execute calls cannot stop the scheduler.
        };
      } else {
        return [&self]() {
          self.stopUnlocked();
        };
      }
    }
 
    void runThreadLoop() {
      std::unique_lock lock {m_mutex};
      while (m_keep_alive) {
        m_syncing_thread = false;
        waitForNextRetry(lock);
 
        if (m_syncing_thread) {
          // Thread was waken up to sync sleep time or to be stopped.
          continue;
        }
 
        try {
          SchedulerStopToken scheduler_stop_token {[this]() {
            clearThreadLoopUnlocked();
          }};
          m_retry_function(*m_iface, scheduler_stop_token);
          continue;
        } catch (const std::exception &error) {  // NOSONAR(cpp:S1181): Scheduler callback boundary must catch standard callback failures.
          detail::logSchedulerException(error, "Exception thrown in the RetryScheduler thread. Stopping scheduler.");
        }
 
        clearThreadLoopUnlocked();
      }
    }
 
    void waitForNextRetry(std::unique_lock<std::mutex> &lock) {
      if (const 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;
        });
        return;
      }
 
      // We're going to sleep until manually woken up.
      m_sleep_cv.wait(lock, [this]() {
        return m_syncing_thread;
      });
    }
 
    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; /* NOSONAR(cpp:S6168): std::jthread is unavailable on the macOS libc++ used by CI. */  
  };
}  // namespace display_device