AsyncJob.h

Go to the documentation of this file.

/*
 * Copyright 2025 CNRS-UM LIRMM, CNRS-AIST JRL
 */
 
#pragma once
#include <mc_rtc/clock.h>
#include <mc_rtc/gui/StateBuilder.h>
#include <mc_rtc/log/Logger.h>
#include <future>
#include <optional>
 
namespace mc_rtc::threading
{
 
template<typename Derived, typename Input, typename Result>
struct AsyncJob
{
  struct Timers
  {
    double dt_startAsync = 0;
    double dt_checkResult = 0;
    double dt_loggerImpl = 0;
    double dt_guiImpl = 0;
 
    // Async loggers
    // Technically not guaranteed to be portable on all plateforms, but should be fine for the ones we care about
    std::atomic<double> dt_compute{0};
  };
 
  AsyncJob() = default;
  AsyncJob(const AsyncJob &) = delete;
  AsyncJob & operator=(const AsyncJob &) = delete;
  AsyncJob(AsyncJob &&) = delete;
  AsyncJob & operator=(AsyncJob &&) = delete;
 
  ~AsyncJob()
  {
    removeFromLogger();
    removeFromGUI();
  }
 
  Input & input()
  {
    if(running_)
    {
      throw std::runtime_error("AsyncJob: cannot get write-access to the input while an async job is running. Please "
                               "ensure that running() is false before modifying the input in place");
    }
    return input_;
  }
 
  void startAsync()
  {
    auto start_async = mc_rtc::clock::now();
    futureResult_ = std::async(
        [this]()
        {
          auto start_compute = mc_rtc::clock::now();
          auto result = derived().computeJob();
          timers_.dt_compute = mc_rtc::elapsed_ms_count(start_compute);
          return result;
        });
    running_ = true;
    startedOnce_ = true;
    canGetSharedState_ = true;
    timers_.dt_startAsync = mc_rtc::elapsed_ms_count(start_async);
  }
 
  bool running() const noexcept { return running_; }
 
  bool startedOnce() const noexcept { return startedOnce_; }
 
  bool checkResult()
  {
    if(!running_) return false;
    auto start_check = mc_rtc::clock::now();
    if(futureResult_.wait_for(std::chrono::seconds(0)) == std::future_status::ready && futureResult_.valid()
       && canGetSharedState_)
    {
 
      try
      {
        lastResult_ = futureResult_.get();
      }
      catch(const std::exception & e)
      {
        mc_rtc::log::error("[{}] Exception in async job: {}", derived().name(), e.what());
        running_ = false;
        canGetSharedState_ = false;
        return false;
      }
 
      canGetSharedState_ = false; // prevent invalid multiple calls to future::get()
      if(logger_ && !inLogger_)
      {
        timers_.dt_loggerImpl = mc_rtc::timedExecution([this]() { addToLogger_(); }).count();
        inLogger_ = true;
      }
      else
      {
        timers_.dt_loggerImpl = 0;
      }
 
      if(gui_ && !inGUI_)
      {
        timers_.dt_guiImpl = mc_rtc::timedExecution([this]() { addToGUI_(); }).count();
        inGUI_ = true;
      }
      else
      {
        timers_.dt_guiImpl = 0;
      }
      running_ = false;
      return true;
    }
    timers_.dt_checkResult = mc_rtc::elapsed_ms_count(start_check);
    return false;
  }
 
  const std::optional<Result> & lastResult() const noexcept { return lastResult_; }
 
  void addToLogger(mc_rtc::Logger & logger, const std::string & prefix)
  {
    if(logger_ && inLogger_) return;
    logger_ = &logger;
    loggerPrefix_ = prefix;
  }
 
  void addToGUI(mc_rtc::gui::StateBuilder & gui, const std::vector<std::string> & category)
  {
    if(gui_ && inGUI_) return;
    gui_ = &gui;
    guiCategory_ = category;
  }
 
  void removeFromLogger()
  {
    if(logger_)
    {
      logger_->removeLogEntries(this);
      inLogger_ = false;
    }
  }
 
  void removeFromGUI()
  {
    if(gui_)
    {
      gui_->removeElements(this);
      inGUI_ = false;
    }
  }
 
  std::string name() const { return "AsyncJob"; }
 
protected: // bookkeeping for the async job
  std::atomic<bool> running_ = false;
  bool inLogger_ = false;
  bool inGUI_ = false;
  bool startedOnce_ = false;
 
  std::future<Result> futureResult_;
  // true if futureResult_.get() has not yet been called, false otherwise
  bool canGetSharedState_ = false;
 
  std::optional<Result> lastResult_;
  Timers timers_;
 
  Input input_;
 
  // CRTP: derived must implement this
  // Result computeJob();
 
  // CRTP: derived may implement these
  void addToLoggerImpl() {}
  void addToGUIImpl() {}
 
  // Store context for logging/GUI
  mc_rtc::Logger * logger_ = nullptr;
  std::string loggerPrefix_ = "";
  mc_rtc::gui::StateBuilder * gui_ = nullptr;
  std::vector<std::string> guiCategory_;
 
protected:
  Derived & derived() { return static_cast<Derived &>(*this); }
 
  void addToLogger_()
  {
    auto contactPrefix = "perf_" + loggerPrefix_ + "_";
    logger_->addLogEntry(contactPrefix + "startAsync [ms]", this, [this]() { return timers_.dt_startAsync; });
    logger_->addLogEntry(contactPrefix + "checkResult [ms]", this, [this]() { return timers_.dt_checkResult; });
    logger_->addLogEntry(contactPrefix + "loggerImpl [ms]", this, [this]() { return timers_.dt_loggerImpl; });
    logger_->addLogEntry(contactPrefix + "guiImpl [ms]", this, [this]() { return timers_.dt_guiImpl; });
    logger_->addLogEntry(contactPrefix + "async_compute [ms]", this, [this]() { return timers_.dt_compute.load(); });
    derived().addToLoggerImpl();
  }
 
  void addToGUI_() { derived().addToGUIImpl(); }
};
 
// Helper alias for CRTP inheritance
template<typename Derived, typename Input, typename Result>
using MakeAsyncJob = AsyncJob<Derived, Input, Result>;
} // namespace mc_rtc::threading