Executor.hxx

Go to the documentation of this file.

 
#ifndef _EXECUTOR_EXECUTOR_HXX_
#define _EXECUTOR_EXECUTOR_HXX_
 
#include <functional>
#include <atomic>
 
#include "executor/Executable.hxx"
#include "executor/Notifiable.hxx"
#include "executor/Selectable.hxx"
#include "executor/Timer.hxx"
#include "utils/Queue.hxx"
#include "utils/SimpleQueue.hxx"
#include "utils/LinkedObject.hxx"
#include "utils/logging.h"
#include "utils/macros.h"
#include "os/OSSelectWakeup.hxx"
 
#ifdef ESP_NONOS
extern "C" {
#include <ets_sys.h>
}
#endif
 
class ActiveTimers;
 
class ExecutorBase : protected OSThread, protected Executable, public LinkedObject<ExecutorBase>
{
public:
    ExecutorBase();
 
    ~ExecutorBase();
 
    static ExecutorBase *by_name(const char *name, bool wait);
 
    virtual void add(Executable *action, unsigned priority = UINT_MAX) = 0;
 
    void sync_run(std::function<void()> fn);
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    virtual void add_from_isr(Executable *action,
                              unsigned priority = UINT_MAX) = 0;
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    void select(Selectable* job);
 
    bool is_selected(Selectable* job);
 
    void unselect(Selectable* job);
 
    bool loop_once();
 
    long long loop_some() ICACHE_FLASH_ATTR;
 
    ActiveTimers* active_timers() { return &activeTimers_; }
 
    void shutdown();
 
    virtual bool empty() = 0;
 
    os_thread_t thread_handle() { return OSThread::get_handle(); }
 
    OSThread& thread() { return *this; }
    
    void assert_current() { HASSERT(os_thread_self() == thread_handle()); }
    
    virtual uint32_t sequence() = 0;
 
    Executable* current() { return current_; }
    
protected:
    void *entry() override;
 
    void run() override {}
 
    OSSelectWakeup selectHelper_;
 
private:
    virtual Executable *next(unsigned *priority) = 0;
 
    void wait_with_select(long long next_timer_nsec);
 
    fd_set *get_select_set(Selectable::SelectType type)
    {
        switch (type)
        {
        case Selectable::READ: return &selectRead_;
        case Selectable::WRITE: return &selectWrite_;
        case Selectable::EXCEPT: return &selectExcept_;
        }
        LOG(FATAL, "Unexpected select type %d", type);
        return nullptr;
    }
 
    const char *name_;
 
    Executable* volatile current_;
 
    ActiveTimers activeTimers_;
 
    fd_set selectRead_;
    fd_set selectWrite_;
    fd_set selectExcept_;
    int selectNFds_;
    TypedQueue<Selectable> selectables_;
 
    std::atomic_uint_least8_t done_;
    std::atomic_uint_least8_t started_;
    unsigned selectPrescaler_ : 5;
 
protected:
    volatile unsigned sequence_ : 25;
    
    friend class Service;
 
    DISALLOW_COPY_AND_ASSIGN(ExecutorBase);
};
 
class NO_THREAD {
public:
    NO_THREAD() {}
};
 
template <unsigned NUM_PRIO>
class Executor : public ExecutorBase
{
public:
 
    Executor(const char *name, int priority, size_t stack_size)
    {
        start_thread(name, priority, stack_size);
    }
 
    explicit Executor(const NO_THREAD& unused) {}
 
    void start_thread(const char *name, int priority, size_t stack_size)
    {
        OSThread::start(name, priority, stack_size);
    }
 
    ~Executor();
 
    void add(Executable *msg, unsigned priority = UINT_MAX) OVERRIDE
    {
        queue_.insert(
            msg, priority >= NUM_PRIO ? NUM_PRIO - 1 : priority);
#ifdef ESP_NONOS
        extern void wakeup_executor(ExecutorBase* executor);
        wakeup_executor(this);
#else
        selectHelper_.wakeup();
#endif
    }
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    void add_from_isr(Executable *msg, unsigned priority = UINT_MAX) override
    {
#ifdef ESP_PLATFORM
        // On the ESP32 we need to call insert instead of insert_locked to
        // ensure that all code paths lock the queue for consistency since
        // this code path is not guaranteed to be protected by a critical
        // section.
        queue_.insert(
            msg, priority >= NUM_PRIO ? NUM_PRIO - 1 : priority);
#else
        queue_.insert_locked(
            msg, priority >= NUM_PRIO ? NUM_PRIO - 1 : priority);
#endif // ESP_PLATFORM
        selectHelper_.wakeup_from_isr();
    }
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    void thread_body() {
        inherit();
    }
 
    bool empty() OVERRIDE
    {
        return queue_.empty();
    }
 
    uint32_t sequence() OVERRIDE { return sequence_; }
 
private:
    Executable *next(unsigned *priority) OVERRIDE
    {
        auto result = queue_.next();
        *priority = result.index;
        return static_cast<Executable*>(result.item);
    }
 
    Executor();
 
    DISALLOW_COPY_AND_ASSIGN(Executor);
 
    QListProtected<NUM_PRIO> queue_;
};
 
class ExecutorGuard : private ::Timer, public SyncNotifiable
{
public:
    ExecutorGuard(ExecutorBase *e)
        : ::Timer(e->active_timers())
        , executor_(e)
    {
        // We wait on the front of the timer queue by expiring immediately.
        start();
    }
 
    long long timeout() override {
        if (executor_->empty()) {
            SyncNotifiable::notify();
            return NONE;
        } else {
            return RESTART;  // wait more on the front of the timer queue
        }
    }
 
private:
    ExecutorBase* executor_;
};
 
template <unsigned NUM_PRIO>
Executor<NUM_PRIO>::~Executor()
{
    shutdown();
}
 
#endif /* _EXECUTOR_EXECUTOR_HXX_ */