loop.hh

Go to the documentation of this file.

 // This Source Code Form is licensed MPL-2.0: http://mozilla.org/MPL/2.0
#pragma once
 
#include <ase/defs.hh>
#include <ase/cotask.hh>
#include <chrono>
#include <mutex>
#include <variant>
 
namespace Ase {
 
// === PollFD ===
struct PollFD   
{
  int           fd;
  uint16        events;
  uint16        revents;
  enum : uint {
    IN          = ASE_SYSVAL_POLLINIT[0],  
    PRI         = ASE_SYSVAL_POLLINIT[1],  
    OUT         = ASE_SYSVAL_POLLINIT[2],  
    RDNORM      = ASE_SYSVAL_POLLINIT[3],  
    RDBAND      = ASE_SYSVAL_POLLINIT[4],  
    WRNORM      = ASE_SYSVAL_POLLINIT[5],  
    WRBAND      = ASE_SYSVAL_POLLINIT[6],  
    // Event types unconditionally updated in .revents
    ERR         = ASE_SYSVAL_POLLINIT[7],  
    HUP         = ASE_SYSVAL_POLLINIT[8],  
    NVAL        = ASE_SYSVAL_POLLINIT[9],  
  };
};
 
// == LoopPriority ==
enum class LoopPriority : uint16_t {
  SYSALLOC = 900, 
  RTAUDIO  = 800, 
  USIGNAL  = 700, 
  COAWAIT  = 600, 
  NOTIFY   = 500, 
  AFRAME   = 400, 
  NORMAL   = 300, 
  IDLE     = 200, 
  LOW      = 100, 
};
 
// === Prototypes ===
class LoopSource;
typedef std::shared_ptr<LoopSource> LoopSourceP;
class TimedSource;
typedef std::shared_ptr<TimedSource> TimedSourceP;
class PollFDSource;
typedef std::shared_ptr<PollFDSource> PollFDSourceP;
class DispatcherSource;
typedef std::shared_ptr<DispatcherSource> DispatcherSourceP;
class USignalSource;
typedef std::shared_ptr<USignalSource> USignalSourceP;
class SigchldSource;
typedef std::shared_ptr<SigchldSource> SigchldSourceP;
class Loop;
typedef std::shared_ptr<Loop> LoopP;
struct LoopState;
#if     defined __G_LIB_H__ || defined DOXYGEN
typedef ::GMainContext GlibGMainContext;
#else
struct GlibGMainContext; // dummy type
#endif
 
// === Concepts ===
template<typename Func>
concept IsLoopCallback = ( std::is_void_v<std::invoke_result_t<Func>> ||
                           std::is_same_v<std::invoke_result_t<Func>, bool> );
 
// === Loop ===
class Loop : public virtual std::enable_shared_from_this<Loop>
{
  friend class LoopImpl;
  ASE_CLASS_NON_COPYABLE (Loop);
protected:
  explicit      Loop                ();
  virtual      ~Loop                ();
  virtual void  destroy_loop        () = 0;
public:
  typedef std::function<void (void)>             VoidSlot;
  typedef std::function<bool (void)>             BoolSlot;
  typedef std::function<void (PollFD&)>          VPfdSlot;
  typedef std::function<bool (PollFD&)>          BPfdSlot;
  typedef std::function<bool (const LoopState&)> DispatcherSlot;
  typedef std::function<bool (int8)>             USignalSlot;
  typedef std::function<void (int,int)>          SigchldSlot;
  template<typename Result> struct Promise;
  virtual void wakeup  () = 0;                
  // source handling
  virtual LoopID add_source    (LoopSourceP loop_source, LoopPriority priority
                                  = LoopPriority::NORMAL) = 0;     
  virtual void cancel          (LoopID            id) = 0;    
  virtual void cancel          (LoopID           *idp) = 0;   
  virtual bool has_primary     (void) = 0;                  
  LoopID exec_dispatcher (const DispatcherSlot &sl, LoopPriority priority
                         = LoopPriority::NORMAL);     
  LoopID exec_usignal    (int8 signum, const USignalSlot &sl, LoopPriority priority
                         = LoopPriority::USIGNAL);     
  virtual LoopID exec_sigchld    (int64_t pid, const SigchldSlot &vfunc, LoopPriority priority
                                  = LoopPriority::NORMAL) = 0;     
  virtual bool exec_once       (uint delay_ms, LoopID *once_id, const VoidSlot &vfunc, LoopPriority priority
                                  = LoopPriority::NORMAL) = 0;     
  template<IsLoopCallback Func>
  LoopID add             (Func &&func, std::chrono::milliseconds interval = std::chrono::milliseconds (0), LoopPriority priority = LoopPriority::NORMAL);
  template<IsLoopCallback Func>
  LoopID add             (Func &&func, LoopPriority priority);
  template<class Coroutine> requires IsAwaitable<std::invoke_result_t<Coroutine>>
  LoopID add             (Coroutine &&coroutine, LoopPriority priority = LoopPriority::NORMAL);
  template<class BoolVoidPollFunctor>
  LoopID exec_io_handler (BoolVoidPollFunctor &&bvf, int fd, const String &mode, LoopPriority priority = LoopPriority::NORMAL);
  // Event processing
  virtual int  run           () = 0; 
  virtual bool running       () = 0; 
  virtual bool finishable    () = 0; 
  virtual void quit          (int quit_code = 0) = 0; 
  virtual bool pending       () = 0; 
  virtual bool iterate       (bool block) = 0; 
  virtual void iterate_pending () = 0; 
  virtual bool set_g_main_context (GlibGMainContext *glib_main_context) = 0; 
  virtual bool has_quit      () = 0; 
  static LoopP current       ();
  template<typename Result>
  std::shared_ptr<Promise<Result>>   make_promise (const std::function<void(std::function<void(Result)>)> &executor);
  std::shared_ptr<Promise<void>>     make_promise (const std::function<void(std::function<void()>)> &executor);
  std::shared_ptr<Promise<uint64_t>> delay        (std::chrono::milliseconds ms);
};
 
// === LoopState ===
struct LoopState {
  enum     Phase { NONE, COLLECT, PREPARE, CHECK, DISPATCH, DESTROY };
  Phase    phase = NONE;
  bool     seen_primary = false;   
  uint64   current_time_usecs = 0; 
  int64    timeout_usecs = 0;      
};
 
// === LoopSource ===
class LoopSource 
{
  friend       class Loop;
  friend       class LoopImpl;
  ASE_CLASS_NON_COPYABLE (LoopSource);
protected:
  Loop   *loop_;
  struct {
    PollFD    *pfd;
    uint       idx;
  }           *pfds_;
  LoopID       id_;
  int16        priority_;
  uint8        loop_state_;
  uint         may_recurse_ : 1;
  uint         dispatching_ : 1;
  uint         was_dispatching_ : 1;
  uint         primary_ : 1;
  uint         n_pfds      ();
  explicit     LoopSource ();
  LoopID       source_id   () { return loop_ ? id_ : LoopID::INVALID; }
  virtual     ~LoopSource ();
public:
  virtual bool prepare     (const LoopState &state,
                            int64 *timeout_usecs_p) = 0;    
  virtual bool check       (const LoopState &state) = 0;    
  virtual bool dispatch    (const LoopState &state) = 0;    
  virtual void destroy     ();
  bool         recursion   () const;                        
  bool         may_recurse () const;                        
  void         may_recurse (bool           may_recurse);    
  bool         primary     () const;                        
  void         primary     (bool           is_primary);     
  void         add_poll    (PollFD * const pfd);            
  void         remove_poll (PollFD * const pfd);            
  void         loop_remove ();                              
  Loop*        loop        () const { return loop_; }       
};
 
// === DispatcherSource ===
class DispatcherSource : public virtual LoopSource 
{
  typedef Loop::DispatcherSlot DispatcherSlot;
  DispatcherSlot slot_;
  ASE_DEFINE_MAKE_SHARED (DispatcherSource);
protected:
  virtual     ~DispatcherSource ();
  virtual bool prepare          (const LoopState &state, int64 *timeout_usecs_p);
  virtual bool check            (const LoopState &state);
  virtual bool dispatch         (const LoopState &state);
  virtual void destroy          ();
  explicit     DispatcherSource (const DispatcherSlot &slot);
public:
  static DispatcherSourceP create (const DispatcherSlot &slot)
  { return make_shared (slot); }
};
 
// === USignalSource ===
class USignalSource : public virtual LoopSource 
{
  typedef Loop::USignalSlot USignalSlot;
  USignalSlot slot_;
  int8        signum_ = 0, index_ = 0, shift_ = 0;
  ASE_DEFINE_MAKE_SHARED (USignalSource);
protected:
  virtual     ~USignalSource  ();
  virtual bool prepare        (const LoopState &state, int64 *timeout_usecs_p);
  virtual bool check          (const LoopState &state);
  virtual bool dispatch       (const LoopState &state);
  virtual void destroy        ();
  explicit     USignalSource  (int8 signum, const USignalSlot &slot);
public:
  static void           raise  (int8 signum);
  static USignalSourceP create (int8 signum, const USignalSlot &slot)
  { return make_shared (signum, slot); }
  static void install_sigaction (int8);
};
 
// === SigchldSource ===
class SigchldSource : public virtual LoopSource 
{
  ASE_DEFINE_MAKE_SHARED (SigchldSource);
  typedef Loop::SigchldSlot SigchldSlot;
  SigchldSlot slot_;
  uint64_t    sigchld_counter_ = 0;
  int64_t     pid_ = 0;
protected:
  virtual     ~SigchldSource   ();
  virtual bool prepare         (const LoopState &state, int64 *timeout_usecs_p);
  virtual bool check           (const LoopState &state);
  virtual bool dispatch        (const LoopState &state);
  virtual void destroy         ();
  explicit     SigchldSource   (int64_t pid, const SigchldSlot &slot);
public:
  static SigchldSourceP create (int64_t pid, const SigchldSlot &slot)
  { return make_shared (pid, slot); }
};
 
// === TimedSource ===
class TimedSource : public virtual LoopSource 
{
  typedef Loop::BoolSlot BoolSlot;
  typedef Loop::VoidSlot VoidSlot;
  uint64     expiration_usecs_;
  uint       interval_msecs_;
  bool       first_interval_;
  const bool oneshot_;
  union {
    BoolSlot bool_slot_;
    VoidSlot void_slot_;
  };
  ASE_DEFINE_MAKE_SHARED (TimedSource);
protected:
  virtual     ~TimedSource  ();
  virtual bool prepare      (const LoopState &state, int64 *timeout_usecs_p);
  virtual bool check        (const LoopState &state);
  virtual bool dispatch     (const LoopState &state);
  explicit     TimedSource  (const BoolSlot &slot, uint initial_interval_msecs, uint repeat_interval_msecs);
  explicit     TimedSource  (const VoidSlot &slot, uint initial_interval_msecs, uint repeat_interval_msecs);
public:
  static TimedSourceP create (const BoolSlot &slot, uint initial_interval_msecs = 0, uint repeat_interval_msecs = 0)
  { return make_shared (slot, initial_interval_msecs, repeat_interval_msecs); }
  static TimedSourceP create (const VoidSlot &slot, uint initial_interval_msecs = 0, uint repeat_interval_msecs = 0)
  { return make_shared (slot, initial_interval_msecs, repeat_interval_msecs); }
};
 
// === PollFDSource ===
class PollFDSource : public virtual LoopSource 
{
  typedef Loop::BPfdSlot BPfdSlot;
  typedef Loop::VPfdSlot VPfdSlot;
protected:
  void          construct       (const String &mode);
  virtual      ~PollFDSource    ();
  virtual bool  prepare         (const LoopState &state, int64 *timeout_usecs_p);
  virtual bool  check           (const LoopState &state);
  virtual bool  dispatch        (const LoopState &state);
  virtual void  destroy         ();
  PollFD        pfd_;
  uint          never_close_ : 1;      // 'C'
private:
  const uint    oneshot_ : 1;
  union {
    BPfdSlot bool_poll_slot_;
    VPfdSlot void_poll_slot_;
  };
  explicit      PollFDSource    (const BPfdSlot &slot, int fd, const String &mode);
  explicit      PollFDSource    (const VPfdSlot &slot, int fd, const String &mode);
  ASE_DEFINE_MAKE_SHARED (PollFDSource);
public:
  static PollFDSourceP create (const BPfdSlot &slot, int fd, const String &mode)
  { return make_shared (slot, fd, mode); }
  static PollFDSourceP create (const VPfdSlot &slot, int fd, const String &mode)
  { return make_shared (slot, fd, mode); }
};
 
// === Loop methods ===
inline LoopID
Loop::exec_dispatcher (const DispatcherSlot &slot, LoopPriority priority)
{
  return add_source (DispatcherSource::create (slot), priority);
}
 
inline LoopID
Loop::exec_usignal (int8 signum, const USignalSlot &slot, LoopPriority priority)
{
  return add_source (USignalSource::create (signum, slot), priority);
}
 
template<class BoolVoidPollFunctor> LoopID
Loop::exec_io_handler (BoolVoidPollFunctor &&bvf, int fd, const String &mode, LoopPriority priority)
{
  using ReturnType = decltype (bvf (*std::declval<PollFD*>()));
  std::function<ReturnType (PollFD&)> slot (bvf);
  return add_source (PollFDSource::create (slot, fd, mode), priority);
}
 
template<IsLoopCallback Func> LoopID
Loop::add (Func &&func, std::chrono::milliseconds interval, LoopPriority priority)
{
  using ReturnType = decltype (func());
  std::function<ReturnType()> slot (std::forward<decltype (func)> (func));
  const uint interval_ms = interval.count();
  return add_source (TimedSource::create (std::move (slot), interval_ms, interval_ms), priority);
}
 
template<IsLoopCallback Func> LoopID
Loop::add (Func &&func, LoopPriority priority)
{
  return add (std::forward<decltype (func)> (func), std::chrono::milliseconds (0), priority);
}
 
template<class Coroutine> requires IsAwaitable<std::invoke_result_t<Coroutine>> LoopID
Loop::add (Coroutine &&coroutine, LoopPriority priority)
{
  using CoroutineType = std::decay_t<Coroutine>;
  if constexpr (std::is_convertible_v<CoroutineType, bool>)
    ASE_ASSERT_RETURN (coroutine, LoopID::INVALID);
  // Keep Coroutine and parameters alive until after co_return
  auto coroutine_ptr = std::make_shared<CoroutineType> (std::forward<Coroutine> (coroutine));
  // Guarantee: coroutine will be started as a loop callback
  auto start_coroutine = [coroutine_ptr] ()
  {
    // dprintf (2, "%s: start loop_coawait…\n", "start_coroutine");
    static auto loop_coawait = [] (auto coroutine_ptr) -> DetachedTask
    {
      // dprintf (2, "%s: co_await…\n", "loop_coawait");
      co_await (*coroutine_ptr) ();
      // dprintf (2, "%s: co_return…\n", "loop_coawait");
      co_return;
    };
    loop_coawait (coroutine_ptr);
    // dprintf (2, "%s: done, return false\n", "start_coroutine");
    return false;
  };
  return this->add (start_coroutine, std::chrono::milliseconds (0), priority);
}
 
// === Loop::Promise ===
template<typename Result>
struct Loop::Promise : std::enable_shared_from_this<Promise<Result>> {
  // If Result is void, we store std::monostate in the 'Resolved' slot
  using StorageType = std::conditional_t<std::is_void_v<Result>, std::monostate, Result>;
  // The State Variant: { Pending | Resolved | Rejected }
  using StateVariant = std::variant<std::monostate, StorageType, std::exception_ptr>;
  struct PromiseAwaiter {
    std::shared_ptr<Promise> promise_;
    bool
    await_ready() const noexcept
    {
      // No lock, state_ is stable once not 0
      return promise_->state_.index() > 0;
    }
    bool
    await_suspend (std::coroutine_handle<> waiter) noexcept
    {
      std::unique_lock lock (promise_->mtx_);
      if (promise_->state_.index() > 0)
        return false; // already resolved
      promise_->waiters_.push_back (waiter);  // caller to resume on resolve()
      return true;    // caller suspended
    }
    Result
    await_resume()
    {
      ASE_ASSERT_WARN (promise_->state_.index() > 0);
      // No lock, state_ is stable once not 0
      if (auto *ex = std::get_if<2> (&promise_->state_))
        ase_rethrow (*ex);
      if constexpr (std::is_void_v<Result>)
        return; // Nothing to return
      else
        // move-only results are empty after first awaiter
        return std::move (std::get<1> (promise_->state_));
    }
  };
  PromiseAwaiter
  operator co_await() noexcept
  {
    return PromiseAwaiter (this->shared_from_this());
  }
 
  template<typename T = Result> requires (!std::is_void_v<T>) void
  resolve (T &&value)
  {
    resolve_impl (StateVariant (std::in_place_index<1>, std::forward<T> (value)));
  }
  template<typename T = Result> requires (std::is_void_v<T>) void
  resolve()
  {
    resolve_impl (StateVariant (std::in_place_index<1>, std::monostate{}));
  }
  void
  reject (std::exception_ptr exc = nullptr)
  {
    if (!exc)
      exc = std::make_exception_ptr (std::runtime_error ("Promise rejected"));
    resolve_impl (StateVariant {std::in_place_index<2>, exc});
  }
  ~Promise() {}
private:
  friend class Loop;
  std::mutex mtx_;
  StateVariant state_;
  Loop *loop_;
  std::vector<std::coroutine_handle<>> waiters_;
  explicit Promise (Loop &loop) : loop_ (&loop) {}
  // State transition to non-pending with new std::variant
  void
  resolve_impl (StateVariant &&new_state)
  {
    std::vector<std::coroutine_handle<>> waiters_to_notify;
    {
      std::lock_guard lock (mtx_);
      if (state_.index() != 0) return;        // Already resolved/rejected
      state_ = std::move (new_state);
      waiters_to_notify.swap (waiters_);
    } // guard released
    auto keep_alive = this->shared_from_this();
    for (auto h : waiters_to_notify)
      if (h && !h.done())
        loop_->add ([h, keep_alive]
        {
          if (!h.done())
            h.resume();
          return false; // one-shot
        }, LoopPriority::COAWAIT);
  }
};
 
template<typename Result>
using LoopPromiseP = std::shared_ptr<Loop::Promise<Result>>;
 
template<typename T> auto
operator co_await (LoopPromiseP<T> p)
{
  return p->operator co_await();
}
 
template<typename Result> std::shared_ptr<Loop::Promise<Result>>
Loop::make_promise (const std::function<void(std::function<void(Result)>)> &executor)
{
  struct EnableMakeShared : Promise<Result> {
    EnableMakeShared (Loop &l) : Promise<Result> (l) {}
  };
  auto promisep = std::make_shared<EnableMakeShared> (*this);
  auto resolver = [promisep] (Result value) { promisep->resolve (value); };
  try {
    executor (resolver);
  } catch (...) {
    promisep->reject (std::current_exception());
  }
  return promisep;
}
 
inline std::shared_ptr<Loop::Promise<void>>
Loop::make_promise (const std::function<void(std::function<void()>)> &executor)
{
  struct EnableMakeShared : Promise<void> {
    EnableMakeShared (Loop &l) : Promise<void> (l) {}
  };
  auto promisep = std::make_shared<EnableMakeShared> (*this);
  auto resolver = [promisep] () { promisep->resolve(); };
  try {
    executor (resolver);
  } catch (...) {
    promisep->reject (std::current_exception());
  }
  return promisep;
}
 
} // Ase