AIStatefulTask.h

Go to the documentation of this file.

 
#ifndef AISTATEFULTASK_H
#define AISTATEFULTASK_H
 
#include "threadpool/Timer.h"
#include "threadsafe/aithreadsafe.h"
#include "threadsafe/AIMutex.h"
#include "threadpool/AIQueueHandle.h"
#include "utils/AIRefCount.h"
#include "utils/macros.h"
#include "utils/FuzzyBool.h"
#include "utils/is_power_of_two.h"
#include "debug.h"
#include <list>
#include <chrono>
#include <functional>
#include <tuple>
#ifdef TRACY_FIBERS
#include <Tracy.hpp>
#include <cstring>
#include <cstdlib>
#endif
 
class AIEngine;
class AIStatefulTaskMutex;
struct AIStatefulTaskMutexNode;
 
using AIWaitConditionFunc = std::function<bool()>;
 
class AIStatefulTask : public AIRefCount
{
 public:
  using state_type = uint32_t;        
  using condition_type = uint32_t;    
  // The two most significant bits are reserved for flow control.
  static constexpr condition_type slow_down_condition = 0x40000000;             // This task was running when the thread pool ran full.
  static constexpr condition_type thread_pool_full_condition = 0x80000000;      // This task couldn't be added to the thread pool queue because that was full.
  static constexpr condition_type AND_conditions_mask = 0xf0000000;
  static constexpr condition_type OR_conditions_mask = 0x0fffffff;
 
  struct Actuation
  {
    boost::intrusive_ptr<AIStatefulTask> task;
    condition_type condition;
 
    void signal() const { task->signal(condition); }
  };
 
 private:
  enum event_type {
    initial_run,              
    schedule_run,             
    normal_run,               
    insert_abort              
  };
 
 protected:
  enum base_state_type {
    bs_reset,                 
    bs_initialize,            
    bs_multiplex,             
    bs_abort,                 
    bs_finish,                
    bs_callback,              
    bs_killed                 
  };
 
 public:
  static constexpr state_type state_end = bs_killed + 1;
  enum on_abort_st {
    abort_parent,             
    signal_parent,            
    do_nothing                
  };
 
  struct Handler
  {
    enum type_t {
      idle_h,           
      immediate_h,      
      engine_h,         
      thread_pool_h     
    };
    enum special_t {
      idle = idle_h,            
      immediate = immediate_h   
    };
    union Handle {
      AIEngine* engine;                 
      AIQueueHandle queue_handle;       
 
      Handle() { }
      Handle(AIEngine* engine_) : engine(engine_) { }
      Handle(AIQueueHandle queue_handle_) : queue_handle(queue_handle_) { }
    };
    Handle m_handle;    
    type_t m_type;      
 
    Handler(special_t special) : m_type((type_t)special) { }
    Handler(AIEngine* engine) : m_handle(engine), m_type(engine_h) { ASSERT(engine); }
    Handler(AIQueueHandle queue_handle) : m_handle(queue_handle), m_type(thread_pool_h) { }
    bool is_engine() const { return m_type == engine_h; }
    bool is_immediate() const { return m_type == immediate_h; }
    bool is_thread_pool() const { return m_type == thread_pool_h; }
    AIQueueHandle get_queue_handle() const { return m_type == thread_pool_h ? m_handle.queue_handle : AIQueueHandle((std::size_t)0); }
    explicit operator bool() const { return m_type != idle_h; }        // Return true when this handler can be used to actually run in.
 
    // Gcc complains because what is defined depends on the value of m_type.
    PRAGMA_DIAGNOSTIC_PUSH_IGNORE_maybe_uninitialized
    bool operator==(Handler handler) const {
        return m_type == handler.m_type &&
            (m_type != engine_h || m_handle.engine == handler.m_handle.engine) &&
            (m_type != thread_pool_h || m_handle.queue_handle == handler.m_handle.queue_handle); }
    PRAGMA_DIAGNOSTIC_POP
    bool operator!=(Handler handler) const { return !operator==(handler); }
 
    friend std::ostream& operator<<(std::ostream& os, Handler const& handler);
  };
 
  struct Conditions
  {
    AIStatefulTask const* m_task;
    condition_type m_conditions;
    Conditions(AIStatefulTask const* task, condition_type conditions) : m_task(task), m_conditions(conditions) { }
 
    void print_on(std::ostream& os) const;
 
    friend std::ostream& operator<<(std::ostream& os, Conditions conditions)
    {
      os << std::hex << conditions.m_conditions << std::dec << " (";
      if (AI_LIKELY(utils::is_power_of_two(conditions.m_conditions)))
        return os << conditions.m_task->condition_str_impl(conditions.m_conditions) << ')';
      conditions.print_on(os);
      return os << ')';
    }
  };
 
  // Also needs access to condition_str_impl.
  friend std::ostream& operator<<(std::ostream&, Conditions);
 
  Conditions print_conditions(condition_type conditions)
  {
    return { this, conditions };
  }
 
 protected:
#ifndef DOXYGEN
  struct multiplex_state_st {
    base_state_type base_state;
    Handler current_handler;            // Current handler.
    AIWaitConditionFunc wait_condition;
    condition_type conditions;
    multiplex_state_st() : base_state(bs_reset), current_handler(Handler::idle), wait_condition(nullptr) { }
  };
 
  struct sub_state_st {
    condition_type skip_wait;         
    condition_type idle;              
    state_type run_state;
    bool reset;
    bool need_run;
    bool aborted;
    bool finished;
#if CW_DEBUG
    bool wait_called;
#endif
  };
 
 private:
  // Base state.
  using multiplex_state_type = aithreadsafe::Wrapper<multiplex_state_st, aithreadsafe::policy::Primitive<std::mutex>>;
  multiplex_state_type mState;
 
 protected:
  // Sub state.
  using sub_state_type = aithreadsafe::Wrapper<sub_state_st, aithreadsafe::policy::Primitive<std::mutex>>;
  sub_state_type mSubState;
#endif // DOXYGEN
 
 private:
  // Mutex protecting everything below and making sure only one thread runs the task at a time.
  AIMutex mMultiplexMutex;
  // Mutex that is locked while calling *_impl() functions and the call back.
  std::recursive_mutex mRunMutex;
 
  using clock_type = std::chrono::steady_clock;
  using duration_type = clock_type::duration;
 
  clock_type::rep mSleep;   
 
  // Callback facilities.
  // From within an other stateful task:
  boost::intrusive_ptr<AIStatefulTask> mParent;       // The parent object that started this task, or nullptr if there isn't any.
  condition_type mParentCondition;                    // The condition (bit) that the parent should be signaled with upon a successful finish.
  on_abort_st mOnAbort;                               // What to do with the parent (if any) when aborted.
  // From outside a stateful task:
  std::function<void (bool)> mCallback;               // Pointer to signal/connection, or nullptr when not connected.
 
  // Engine stuff.
  Handler mDefaultHandler;            // Default engine or queue.
  Handler mTargetHandler;             // Requested engine by a call to yield.
  bool mYield;                        // True when any yield function was called, except for yield_if_not when the passed engine already matched.
 
#if CW_DEBUG
  // Debug stuff.
  std::thread::id mThreadId;          // The thread currently running multiplex() (or std::thread::id() when none).
  base_state_type mDebugLastState;    // The previous state that multiplex() had a normal run with.
  bool mDebugShouldRun;               // Set if we found evidence that we should indeed call multiplex_impl().
  bool mDebugAborted;                 // True when abort() was called.
  bool mDebugSignalPending;           // True while wait() was called but didn't get idle because of a pending call to signal() that wasn't handled yet.
  bool mDebugSetStatePending;         // True while set_state() was called by not handled yet.
  bool mDebugRefCalled;               // True when ref() is called (or will be called within the critial area of mMultiplexMutex).
#endif
 
  static thread_local AIStatefulTask* tl_parent_task;
 
#if defined(CWDEBUG) && !defined(DOXYGEN)
 protected:
  bool mSMDebug;                      // Print debug output only when true (SM = 'State Machine', the name of this class before it was renamed to StatefulTask).
#endif
#if CW_DEBUG && !defined(DOXYGEN)
 public:
  bool m_may_not_be_deleted;
#endif
 
 private:
  duration_type mDuration;            // Total time spent running in the main thread.
 
#ifdef TRACY_FIBERS
 protected:
  char const* m_tracy_fiber_name;     // Set by call to set_tracy_fiber_name. Normally equal to the return value of task_name_impl()
                                      // called directly after creation (i.e. create or create_from_tuple).
                                      // Made protected to allow a derived class to change the name dynamically: this should happen
                                      // from multiplex_impl only (otherwise it is not thread-safe) and will change the name for
                                      // the *next* invocation of multiplex(). The actual fiber is determined by the value of the
                                      // pointer, not the string that it points to.
  static thread_local char const* s_tl_tracy_fiber_name;  // Set to m_tracy_fiber_name when entering multiplex().
#endif
 
 public:
  AIStatefulTask(CWDEBUG_ONLY(bool debug)) : mDefaultHandler(Handler::idle), mTargetHandler(Handler::idle), mYield(false),
#if CW_DEBUG
  mDebugLastState(bs_killed), mDebugShouldRun(false), mDebugAborted(false), mDebugSignalPending(false),
  mDebugSetStatePending(false), mDebugRefCalled(false),
#endif
#ifdef CWDEBUG
  mSMDebug(debug),
#endif
#if CW_DEBUG
  m_may_not_be_deleted(false),
#endif
  mDuration(duration_type::zero())
#ifdef TRACY_FIBERS
  , m_tracy_fiber_name(nullptr)
#endif
  { }
 
#ifdef TRACY_FIBERS
  void set_tracy_fiber_name(char const* tracy_fiber_name);
#endif
 
 protected:
  virtual ~AIStatefulTask()
  {
#if CW_DEBUG
    base_state_type state = multiplex_state_type::rat(mState)->base_state;
    ASSERT(state == bs_killed || state == bs_reset);
    ASSERT(!m_may_not_be_deleted);
#endif
  }
 
 public:
  void run(Handler default_handler, std::function<void (bool)> cb_function);
 
  void run(std::function<void (bool)> cb_function) { run(Handler::immediate, cb_function); }
 
  void run(Handler default_handler, AIStatefulTask* parent, condition_type condition, on_abort_st on_abort = abort_parent);
 
  void run(AIStatefulTask* parent, condition_type condition, on_abort_st on_abort = abort_parent) { run(Handler::immediate, parent, condition, on_abort); }
 
  void run(Handler default_handler = Handler::immediate) { run(default_handler, nullptr, 0, do_nothing); }
 
 
  void kill();
 
 protected:
  void set_state(state_type new_state);       // Run this state the NEXT loop.
 
  void wait(condition_type conditions);
 
  void wait_until(AIWaitConditionFunc const& wait_condition, condition_type conditions);
 
  void wait_until(AIWaitConditionFunc const& wait_condition, condition_type conditions, state_type new_state)
  {
    set_state(new_state);
    wait_until(wait_condition, conditions);
  }
 
 
  void finish();
 
  inline utils::FuzzyBool is_self_locked(AIStatefulTaskMutex const& stateful_task_mutex, AIStatefulTaskMutexNode const* handle);
 
  void yield();
 
  void target(Handler handler);
 
  void yield(Handler handler);
 
  void yield_frame(AIEngine* engine, unsigned int frames);
 
  void yield_ms(AIEngine* engine, unsigned int ms);
 
  bool yield_if_not(Handler handler);
 
 
  // Calls wait_until.
  friend class AIFriendOfStatefulTask;
 
 public:
  void abort();
 
  bool signal(condition_type condition);
 
 
 public:
  // Accessors.
 
  bool running() const { return multiplex_state_type::crat(mState)->base_state == bs_multiplex; }
 
  bool waiting() const;
 
  bool waiting_or_aborting() const;
 
  bool active(Handler handler) const { return multiplex_state_type::crat(mState)->current_handler == handler; }
 
  bool is_immediate() const { return multiplex_state_type::crat(mState)->current_handler.is_immediate(); }
 
  // For debugging purposes mainly.
  bool default_is_immediate() const { return mDefaultHandler.is_immediate(); }
 
  bool finished() const
  {
    sub_state_type::crat sub_state_r(mSubState);
    return sub_state_r->finished ? &AIStatefulTask::mOnAbort : 0;
  }
 
  bool aborted() const { return sub_state_type::crat(mSubState)->aborted; }
 
  bool executing() const { return mMultiplexMutex.is_self_locked(); }
 
  static char const* state_str(base_state_type state);
#ifdef CWDEBUG
  static char const* event_str(event_type event);
#endif
 
  duration_type getDuration() const { return mDuration; }
 
  char const* task_name() const { return task_name_impl(); }
 
 protected:
  virtual char const* condition_str_impl(condition_type condition) const;
  virtual char const* state_str_impl(state_type run_state) const;
  virtual char const* task_name_impl() const = 0;
  virtual void initialize_impl();
  virtual void multiplex_impl(state_type run_state) = 0;
  virtual void abort_impl();
  virtual void finish_impl();
  virtual void force_killed();
 
 private:
  void reset();                                       // Called from run() to (re)initialize a (re)start.
 
  void multiplex(event_type event, Handler handler = Handler::idle);
 
  void insert_multiplex(event_type event, Handler handler = Handler::idle)
  {
#ifdef TRACY_FIBERS
    char const* parent_tracy_fiber_name = s_tl_tracy_fiber_name;
    if (AI_UNLIKELY(parent_tracy_fiber_name))
    {
      //Dout(dc::notice, "Calling TracyFiberLeave()");
      TracyFiberLeave;
      s_tl_tracy_fiber_name = nullptr;
    }
#endif
    multiplex(event, handler);
#ifdef TRACY_FIBERS
    if (AI_UNLIKELY(parent_tracy_fiber_name))
    {
      s_tl_tracy_fiber_name = parent_tracy_fiber_name;
      //Dout(dc::notice, "Calling TracyFiberEnter(\"" << parent_tracy_fiber_name << "\")");
      TracyFiberEnter(parent_tracy_fiber_name);
    }
#endif
  }
 
  state_type begin_loop();                            // Called from multiplex() at the start of a loop.
  void callback();                                    // Called when the task finished.
  // Count frames if necessary and return true when the task is still sleeping.
  // Sleeping in milliseconds (as opposed to counting frames) is assumed to only be done in order
  // to give other tasks cpu. Therefore, if 'only_task' is set and this sleep is not counting
  // frames, then stop sleeping.
  bool sleep(clock_type::time_point current_time, bool only_task)
  {
    if (mSleep == 0)
      return false;
    else if (mSleep < 0)
      ++mSleep;
    else if (mSleep <= current_time.time_since_epoch().count() || only_task)
      mSleep = 0;
    return mSleep != 0;
  }
 
  void add(duration_type delta) { mDuration += delta; }
 
  void add_task_to_thread_pool(AIQueueHandle queue_handle, uint8_t const failure_count = 0);    // Attempt to add this task to a theadpool queue.
  void wait_AND(condition_type required);                                                       // Stop running until all `required` bits have been signaled (plus at least one of any other wait() condition).
 
  friend class AIEngine;      // Calls multiplex(), force_killed() and add().
};
 
namespace task {
 
template<typename Type>
concept TaskType = std::is_base_of_v<AIStatefulTask, Type>;
 
} // namespace task
 
#if defined(CWDEBUG) && !defined(DOXYGEN)
NAMESPACE_DEBUG_CHANNELS_START
extern channel_ct statefultask;
NAMESPACE_DEBUG_CHANNELS_END
#endif
 
namespace statefultask {
 
template<task::TaskType TaskType, typename... ARGS>
boost::intrusive_ptr<TaskType> create(ARGS&&... args)
{
  DoutEntering(dc::statefultask, "statefultask::create<" << ::NAMESPACE_DEBUG::type_name_of<TaskType>() <<
      ((LibcwDoutStream << ... << (std::string(", ") + ::NAMESPACE_DEBUG::type_name_of<ARGS>())), ">(") << join(", ", args...) << ')');
  TaskType* task = new TaskType(std::forward<ARGS>(args)...);
  AllocTag2(task, "Created with statefultask::create");
#ifdef TRACY_FIBERS
  task->set_tracy_fiber_name(task->task_name());
#endif
#ifdef CWDEBUG
  Dout(dc::statefultask|continued_cf, "Returning task pointer " << (void*)task);
  if ((void*)task != (void*)static_cast<AIStatefulTask*>(task))
    Dout(dc::finish, " [" << static_cast<AIStatefulTask*>(task) << "].");
  else
    Dout(dc::finish, ".");
#endif
  return task;
}
 
template<task::TaskType TaskType, typename... ARGS>
boost::intrusive_ptr<TaskType> create_from_tuple(std::tuple<ARGS...>&& args)
{
  DoutEntering(dc::statefultask, "statefultask::create_from_tuple<" << ::NAMESPACE_DEBUG::type_name_of<TaskType>() <<
      ((LibcwDoutStream << ... << (std::string(", ") + ::NAMESPACE_DEBUG::type_name_of<ARGS>())), ">(") << args << ')');
  TaskType* task = new TaskType(std::make_from_tuple<TaskType>(std::move(args)));
  AllocTag2(task, "Created with statefultask::create_from_tuple");
#ifdef TRACY_FIBERS
  task->set_tracy_fiber_name(task->task_name());
#endif
#ifdef CWDEBUG
  Dout(dc::statefultask|continued_cf, "Returning task pointer " << (void*)task);
  if ((void*)task != (void*)static_cast<AIStatefulTask*>(task))
    Dout(dc::finish, " [" << static_cast<AIStatefulTask*>(task) << "].");
  else
    Dout(dc::finish, '.');
#endif
  return task;
}
 
} // namespace statefultask
 
#include "AIStatefulTaskMutex.h"
#endif // AISTATEFULTASK_H
 
#ifndef AISTATEFULTASK_H_definitions
#define AISTATEFULTASK_H_definitions
 
utils::FuzzyBool AIStatefulTask::is_self_locked(AIStatefulTaskMutex const& stateful_task_mutex, AIStatefulTaskMutexNode const* handle)
{
  return stateful_task_mutex.is_self_locked(handle);
}
 
#endif // AISTATEFULTASK_H_definitions