StateFlow.hxx

Go to the documentation of this file.

 
#ifndef _EXECUTOR_STATEFLOW_HXX_
#define _EXECUTOR_STATEFLOW_HXX_
 
#include <unistd.h>
#include <type_traits>
#include <functional>
#include <sys/stat.h>
 
#include "executor/Service.hxx"
#include "executor/Timer.hxx"
#include "utils/Buffer.hxx"
#include "utils/Queue.hxx"
#include "utils/LinkedObject.hxx"
 
#define STATE(_fn)                                                             \
    (StateFlowBase::Callback)(                                                 \
        &std::remove_reference<decltype(*this)>::type::_fn)
 
#define STATE_FLOW_STATE(_state) Action _state()
 
#define STATE_FLOW_START(_name, _message, _priorities)                         \
    class _name : public StateFlow<_message, _priorities>                      \
    {                                                                          \
    public:                                                                    \
        _name(Service *service) : StateFlow<_priorities, _priorities>(service) \
        {                                                                      \
        }                                                                      \
                                                                               \
    private:                                                                   \
        STATE_FLOW_STATE(entry);                                               \
                                                                               \
        _name();                                                               \
                                                                               \
    DISALLOW_COPY_AND_ASSIGN(_name)
 
#define STATE_FLOW_START_WITH_TIMER(_name, _priorities)                        \
    class _name : public StateFlow<_priorities>                                \
    {                                                                          \
    public:                                                                    \
        _name(Service *service)                                                \
            : StateFlow<_priorities>(service)                                  \
            , timer(TIMEOUT_FROM(service, state_flow_timeout), service, this)  \
            , timerMsg(NULL)                                                   \
        {                                                                      \
        }                                                                      \
                                                                               \
        ~_name()                                                               \
        {                                                                      \
        }                                                                      \
                                                                               \
        void timeout()                                                         \
        {                                                                      \
            timerMsg ? me()->send(timerMsg) :;                                 \
            timerMsg = NULL;                                                   \
        }                                                                      \
                                                                               \
        void trigger()                                                         \
        {                                                                      \
            timer.trigger();                                                   \
        }                                                                      \
                                                                               \
    private:                                                                   \
        STATE_FLOW_STATE(entry);                                               \
                                                                               \
        Action timeout_and_call(Callback c, Message *msg, long long period)    \
        {                                                                      \
            msg->id(msg->id() | Message::IN_PROCESS_MSK);                      \
            timerMsg = msg;                                                    \
            timer.start(period);                                               \
            return Action(c);                                                  \
        }                                                                      \
                                                                               \
        Timer timer;                                                           \
        Message *timerMsg;                                                     \
                                                                               \
        bool early()                                                           \
        {                                                                      \
            return timer.early();                                              \
        }                                                                      \
                                                                               \
        _name();                                                               \
                                                                               \
    DISALLOW_COPY_AND_ASSIGN(_name)
 
#define STATE_FLOW_END() }
 
template <class T> class FlowInterface;
 
class StateFlowTimer;
 
class StateFlowBase : public Executable
{
public:
    void run() override;
 
    void notify() override;
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    virtual void notify_from_isr() OVERRIDE;
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    Service *service()
    {
        return service_;
    }
 
protected:
    StateFlowBase(Service *service)
        : service_(service)
        , state_(STATE(terminated))
        , allocationResult_(nullptr)
    {
    }
 
    ~StateFlowBase()
    {
    }
 
    /* forward prototype */
    class Action;
 
    typedef Action (StateFlowBase::*Callback)();
 
    class Action
    {
    public:
        Action(Callback s) : nextState_(s)
        {
        }
 
        Callback next_state()
        {
            return nextState_;
        }
 
    private:
        Callback nextState_;
    };
 
    void reset_flow(Callback c)
    {
        state_ = c;
    }
 
    bool is_state(Callback c)
    {
        return state_ == c;
    }
 
    bool is_terminated()
    {
        return is_state(STATE(terminated));
    }
 
    void start_flow(Callback c)
    {
        HASSERT(is_terminated());
        yield_and_call(c);
    }
 
    /*========== ACTION COMMANDS ===============*/
    /* StateFlow implementations will have to use one of the following commands
     * to return from a state handler to indicate what action to take. */
 
    Action again()
    {
        return Action(state_);
    }
 
    Action exit()
    {
        return STATE(terminated);
    }
 
    Action delete_this()
    {
        state_ = STATE(terminated);
        delete this;
        // Ensures that Run() does not touch the class member variables
        // anymore.
        return wait();
    }
 
    Action set_terminated() {
        state_ = STATE(terminated);
        return wait();
    }
 
    Action call_immediately(Callback c)
    {
        return Action(c);
    }
 
    Action wait()
    {
        return Action(nullptr);
    }
 
    Action wait_and_call(Callback c)
    {
        state_ = c;
        return wait();
    }
 
    template <class T>
    Action allocate_and_call(FlowInterface<Buffer<T>> *target_flow, Callback c,
                             Pool *pool = nullptr)
    {
        allocationResult_ = nullptr;
        Pool *p = pool;
        if (!p)
        {
            HASSERT(target_flow != nullptr);
            p = target_flow->pool();
        }
        LOG(VERBOSE, "allocate from pool %p, main pool %p", p, mainBufferPool);
        p->alloc_async<T>(this);
        return wait_and_call(c);
    }
 
    Action allocate_and_call(Callback c, QAsync *queue)
    {
        allocationResult_ = nullptr;
        queue->next_async(this);
        return wait_and_call(c);
    }
 
    template <class T>
    Buffer<T> *full_allocation_result(FlowInterface<Buffer<T>> *target_flow)
    {
        Buffer<T> *result = static_cast<Buffer<T> *>(allocationResult_);
        return result;
    }
 
    template <class T>
    T *full_allocation_result(TypedQAsync<T> *queue)
    {
        T *result = static_cast<T*>(allocationResult_);
        return result;
    }
 
    template <class T>
    void cast_allocation_result(T** member)
    {
        *member = static_cast<T*>(allocationResult_);
    }
 
    template <class T>
    inline Buffer<T> *
    get_allocation_result(FlowInterface<Buffer<T>> *target_flow);
 
    Action yield_and_call(Callback c)
    {
        state_ = c;
        notify();
        return wait();
    }
 
    Action yield()
    {
        notify();
        return wait();
    }
 
    class StateFlowTimer : public ::Timer
    {
    public:
        StateFlowTimer(StateFlowBase *parent)
            : Timer(parent->service()->executor()->active_timers())
            , parent_(parent)
        {
        }
 
        long long timeout() override
        {
            parent_->notify();
            return NONE;
        }
 
    protected:
        StateFlowBase *parent_;
    };
 
    Action sleep_and_call(::Timer *timer, long long timeout_nsec, Callback c)
    {
        timer->start(timeout_nsec);
        return wait_and_call(c);
    }
 
    template <class T, typename... Args>
    Action invoke_subflow_and_wait(
        FlowInterface<Buffer<T>> *target_flow, Callback c, Args &&... args)
    {
        Buffer<T> *b;
        mainBufferPool->alloc(&b);
        b->data()->reset(std::forward<Args>(args)...);
        b->data()->done.reset(this);
        allocationResult_ = b->ref();
        target_flow->send(b);
        return wait_and_call(c);
    }
 
public:
    template <class T, typename... Args>
    static void invoke_subflow_and_ignore_result(
        FlowInterface<Buffer<T>> *target_flow, Args &&... args)
    {
        Buffer<T> *b;
        mainBufferPool->alloc(&b);
        b->data()->reset(std::forward<Args>(args)...);
        b->data()->done.reset(EmptyNotifiable::DefaultInstance());
        target_flow->send(b);
    }
 
protected:
    struct StateFlowSelectHelper;
    struct StateFlowTimedSelectHelper;
 
    Action read_repeated(StateFlowSelectHelper* helper, int fd, void* buf, size_t size, Callback c, unsigned priority = Selectable::MAX_PRIO) {
        helper->reset(Selectable::READ, fd, priority);
        helper->set_wakeup(this);
        helper->rbuf_ = static_cast<uint8_t*>(buf);
        helper->remaining_ = size;
        helper->readFully_ = 1;
        helper->readNonblocking_ = 0;
        helper->readWithTimeout_ = 0;
        helper->nextState_ = c;
        helper->hasError_ = 0;
        allocationResult_ = helper;
        return call_immediately(STATE(internal_try_read));
    }
 
    Action read_single(StateFlowSelectHelper *helper, int fd, void *buf,
        size_t size, Callback c, unsigned priority = Selectable::MAX_PRIO)
    {
        helper->reset(Selectable::READ, fd, priority);
        helper->set_wakeup(this);
        helper->rbuf_ = static_cast<uint8_t*>(buf);
        helper->remaining_ = size;
        helper->readFully_ = 0;
        helper->readNonblocking_ = 0;
        helper->readWithTimeout_ = 0;
        helper->nextState_ = c;
        helper->hasError_ = 0;
        allocationResult_ = helper;
        return call_immediately(STATE(internal_try_read));
    }
 
    Action read_nonblocking(StateFlowSelectHelper* helper, int fd, void* buf, size_t size, Callback c, unsigned priority = Selectable::MAX_PRIO) {
        helper->reset(Selectable::READ, fd, priority);
        helper->set_wakeup(this);
        helper->rbuf_ = static_cast<uint8_t*>(buf);
        helper->remaining_ = size;
        helper->readFully_ = 0;
        helper->readNonblocking_ = 1;
        helper->readWithTimeout_ = 0;
        helper->nextState_ = c;
        helper->hasError_ = 0;
        allocationResult_ = helper;
        return call_immediately(STATE(internal_try_read));
    }
 
    Action read_repeated_with_timeout(StateFlowTimedSelectHelper *helper,
        long long timeout_nsec, int fd, void *buf, size_t size, Callback c,
        unsigned priority = Selectable::MAX_PRIO)
    {
        helper->reset(Selectable::READ, fd, priority);
        helper->set_timed_wakeup();
        helper->rbuf_ = static_cast<uint8_t*>(buf);
        helper->remaining_ = size;
        helper->expiry_ = OSTime::get_monotonic() + timeout_nsec;
        helper->readFully_ = 1;
        helper->readNonblocking_ = 0;
        helper->readWithTimeout_ = 1;
        helper->timer_.set_triggered(); // Needed for the first iteration
        helper->nextState_ = c;
        helper->hasError_ = 0;
        allocationResult_ = static_cast<StateFlowSelectHelper *>(helper);
        return call_immediately(STATE(internal_try_read));
    }
 
    Action internal_try_read()
    {
        StateFlowSelectHelper *h =
            static_cast<StateFlowSelectHelper *>(allocationResult_);
        if (h->readWithTimeout_)
        {
            if (service()->executor()->is_selected(h))
            {
                service()->executor()->unselect(h);
            }
        }
        if (!h->remaining_)
        {
            h->rbuf_ = nullptr;
            return call_immediately(h->nextState_);
        }
        int count = ::read(h->fd(), h->rbuf_, h->remaining_);
        if (count > 0)
        {
            h->remaining_ -= count;
            h->rbuf_ += count;
            if (h->remaining_ && h->readFully_ && !h->readNonblocking_)
            {
                return again();
            }
            else
            {
                h->rbuf_ = nullptr;
                return call_immediately(h->nextState_);
            }
        }
        if (count < 0 &&
            (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
        {
            if (h->readNonblocking_)
            {
                h->rbuf_ = nullptr;
                return call_immediately(h->nextState_);
            }
            else
            {
                // Blocked.
                if (h->readWithTimeout_)
                {
                    auto *hh = static_cast<StateFlowTimedSelectHelper *>(h);
                    if (!hh->timer_.is_triggered())
                    {
                        // We actually got a timeout notification.
                        h->rbuf_ = nullptr;
                        return call_immediately(h->nextState_);
                    }
                    hh->timer_.start_absolute(hh->expiry_);
                }
                service()->executor()->select(h);
                return wait();
            }
        }
        // Now: we are at an unknown error or EOF.
        h->rbuf_ = nullptr;
        h->hasError_ = 1;
        return call_immediately(h->nextState_);
    }
 
 
#if OPENMRN_FEATURE_BSD_SOCKETS
    Action listen_and_call(StateFlowSelectHelper *helper, int fd, Callback c)
    {
#if OPENMRN_HAVE_SOCKET_FSTAT
        // verify that the fd is a socket
        struct stat stat;
        fstat(fd, &stat);
        HASSERT(S_ISSOCK(stat.st_mode));
#endif // OPENMRN_HAVE_SOCKET_FSTAT
 
        helper->reset(Selectable::READ, fd, Selectable::MAX_PRIO);
        helper->set_wakeup(this);
 
        service()->executor()->select(helper);
        return wait_and_call(c);
    }
 
    Action connect_and_call(StateFlowSelectHelper *helper, int fd, Callback c)
    {
#if OPENMRN_HAVE_SOCKET_FSTAT
        // verify that the fd is a socket
        struct stat stat;
        fstat(fd, &stat);
        HASSERT(S_ISSOCK(stat.st_mode));
#endif // OPENMRN_HAVE_SOCKET_FSTAT
 
        helper->reset(Selectable::WRITE, fd, Selectable::MAX_PRIO);
        helper->set_wakeup(this);
 
        service()->executor()->select(helper);
        return wait_and_call(c);
    }
#endif // OPENMRN_FEATURE_BSD_SOCKETS
 
    Action write_repeated(StateFlowSelectHelper *helper, int fd,
        const void *buf, size_t size, Callback c,
        unsigned priority = Selectable::MAX_PRIO)
    {
        helper->reset(Selectable::WRITE, fd, priority);
        helper->set_wakeup(this);
        helper->wbuf_ = static_cast<const uint8_t*>(buf);
        helper->remaining_ = size;
        helper->readFully_ = 1;
        helper->readWithTimeout_ = 0;
        helper->nextState_ = c;
        helper->hasError_ = 0;
        allocationResult_ = helper;
        return call_immediately(STATE(internal_try_write));
    }
 
    Action internal_try_write()
    {
        StateFlowSelectHelper *h =
            static_cast<StateFlowSelectHelper *>(allocationResult_);
        if (!h->remaining_)
        {
            return call_immediately(h->nextState_);
        }
        int count = ::write(h->fd(), h->wbuf_, h->remaining_);
        if (count > 0)
        {
            h->remaining_ -= count;
            h->wbuf_ += count;
            return again();
        }
        if (count <= 0 &&
            (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR))
        {
            // Blocked.
            service()->executor()->select(h);
            return wait();
        }
        h->hasError_ = 1;
#ifdef STATEFLOW_DEBUG_WRITE_ERRORS
        static volatile int scount;
        static volatile int serrno;
        scount = count;
        serrno = errno;
        LOG(FATAL, "failed to write count=%d errno=%d", scount, serrno);
        DIE("failed write");
#endif
        // Now: we are at an unknown error or EOF.
        return call_immediately(h->nextState_);
    }
 
    struct StateFlowSelectHelper : public Selectable
    {
        StateFlowSelectHelper(StateFlowBase *parent)
            : Selectable(parent)
            , hasError_(0)
        {
        }
 
        union
        {
            const uint8_t *wbuf_;
            uint8_t *rbuf_;
        };
 
        Callback nextState_;
        unsigned readFully_ : 1;
        unsigned readNonblocking_ : 1;
        unsigned readWithTimeout_ : 1;
        unsigned hasError_ : 1;
        unsigned remaining_ : 28;
    };
 
    struct StateFlowTimedSelectHelper : public StateFlowSelectHelper,
                                        private Executable
    {
        StateFlowTimedSelectHelper(StateFlowBase *parent)
            : StateFlowSelectHelper(parent)
            , timer_(parent)
        {
        }
 
        void set_timed_wakeup() {
            set_wakeup((Executable*)this);
        }
 
        StateFlowTimer timer_;
 
        long long expiry_;
 
    private:
        // Executable interface. Called by select.
        void run() override {
            timer_.ensure_triggered();
        }
    };
 
private:
    Service *service_;
 
    Action terminated();
 
    void alloc_result(QMember *b) override
    {
        allocationResult_ = b;
        notify();
    }
 
    Callback state_;
 
    QMember *allocationResult_;
 
    StateFlowBase();
 
    DISALLOW_COPY_AND_ASSIGN(StateFlowBase);
};
 
template <class T, class S> class StateFlow;
 
class StateFlowWithQueue : public StateFlowBase, protected Atomic, public LinkedObject<StateFlowWithQueue>
{
public:
    ~StateFlowWithQueue();
 
    void notify() override;
 
#if OPENMRN_FEATURE_RTOS_FROM_ISR
    void notify_from_isr() OVERRIDE;
#endif // OPENMRN_FEATURE_RTOS_FROM_ISR
 
    bool is_waiting()
    {
        AtomicHolder h(this);
        if (!queue_empty()) return false;
        return isWaiting_;
    }
 
protected:
    StateFlowWithQueue(Service *service);
 
    virtual Action entry() = 0;
 
    virtual QMember *queue_next(unsigned *priority) = 0;
 
    virtual bool queue_empty() = 0;
 
    virtual void release() = 0;
 
    Action exit()
    {
        return call_immediately(STATE(wait_for_message));
    }
 
    Action release_and_exit()
    {
        release();
        return exit();
    }
 
    BufferBase *message()
    {
        return currentMessage_;
    }
 
    BufferBase *transfer_message()
    {
        BufferBase *m = message();
        currentMessage_ = nullptr;
        return m;
    }
 
    void reset_message(BufferBase* message, unsigned priority) {
        HASSERT(!currentMessage_);
        currentMessage_ = message;
        set_priority(priority);
    }
 
    unsigned priority()
    {
        return currentPriority_;
    }
 
    void set_priority(unsigned priority)
    {
        currentPriority_ = std::min(priority, MAX_PRIORITY_);
    }
 
    void start_flow_at_init(Callback c)
    {
        reset_flow(c);
        notify();
        isWaiting_ = 0;
    }
 
private:
    STATE_FLOW_STATE(wait_for_message);
 
    unsigned queueSize_;
 
    BufferBase *currentMessage_;
 
    unsigned currentPriority_ : 31;
 
    unsigned isWaiting_ : 1;
 
    template <class Q> friend class UntypedStateFlow;
    template <class M, class B> friend class TypedStateFlow;
    friend class GlobalEventFlow;
 
    static const unsigned MAX_PRIORITY_ = 0x7FFFFFFFU;
};
 
template <class MessageType> class FlowInterface;
 
template <class MessageType> class FlowInterface
{
public:
    typedef MessageType message_type;
 
    virtual ~FlowInterface() {}
 
    virtual Pool *pool()
    {
        return mainBufferPool;
    }
 
    virtual void send(MessageType *message, unsigned priority = UINT_MAX) = 0;
 
    virtual MessageType *type_helper()
    {
        return nullptr;
    }
 
    MessageType *alloc()
    {
        MessageType *ret;
        pool()->alloc(&ret);
        return ret;
    }
 
    void alloc_async(Executable *target)
    {
        typedef typename MessageType::value_type T;
        Pool* p = pool();
        p->alloc_async<T>(target);
    }
 
    static MessageType *cast_alloc(QMember *entry)
    {
        MessageType *result;
        Pool::alloc_async_init(static_cast<BufferBase *>(entry), &result);
        return result;
    }
 
    class GenericHandler;
};
 
template <class MessageType>
class FlowInterface<MessageType>::GenericHandler
    : public FlowInterface<MessageType>
{
public:
    typedef std::function<void(message_type *)> HandlerFn;
    GenericHandler(HandlerFn handler)
        : handler_(handler)
    {
    }
 
    template<class T>
    GenericHandler(T* ptr, void (T::*fn)(message_type*))
        : handler_(std::bind(fn, ptr, std::placeholders::_1)) {}
 
    void send(MessageType *message, unsigned priority) OVERRIDE
    {
        handler_(message);
    }
 
private:
    HandlerFn handler_;
};
 
template <class T>
Buffer<T> *
StateFlowBase::get_allocation_result(FlowInterface<Buffer<T>> *target_flow)
{
    return target_flow->cast_alloc(allocationResult_);
}
 
 
template<class QueueType>
class UntypedStateFlow : public StateFlowWithQueue {
public:
    UntypedStateFlow(Service* service) : StateFlowWithQueue(service) {}
 
    ~UntypedStateFlow()
    {
    }
 
protected:
    void send(BufferBase *msg, unsigned priority = UINT_MAX)
    {
        AtomicHolder h(this);
        queue_.insert_locked(msg, priority);
        queueSize_ = queue_.size();
        if (isWaiting_)
        {
            isWaiting_ = 0;
            set_priority(priority);
            this->notify();
        }
    }
 
    using StateFlowBase::call_immediately;
    using StateFlowBase::Callback;
 
    QMember *queue_next(unsigned *priority) OVERRIDE
    {
        typename QueueType::Result r = queue_.next_locked();
        if (r.item)
        {
            *priority = r.index;
        }
        return r.item;
    }
 
    bool queue_empty() OVERRIDE {
        AtomicHolder h(this);
        return queue_.empty();
    }
 
private:
    QueueType queue_;
};
 
template <class MessageType, class Base>
class TypedStateFlow : public Base, public FlowInterface<MessageType>
{
public:
    typedef typename Base::Action Action;
 
    TypedStateFlow(Service *service) : Base(service) {}
 
    virtual ~TypedStateFlow()
    {
    }
 
    void send(MessageType *msg, unsigned priority = UINT_MAX) OVERRIDE
    {
        Base::send(msg, priority);
    }
 
    virtual Action entry() override = 0;
 
protected:
    void release() OVERRIDE
    {
        if (message())
        {
            message()->unref();
        }
        this->currentMessage_ = nullptr;
    }
 
    void return_buffer()
    {
        message()->data()->done.notify();
        release();
    }
 
    MessageType *message()
    {
        return static_cast<MessageType *>(Base::message());
    }
 
    MessageType *transfer_message()
    {
        return static_cast<MessageType *>(
            Base::transfer_message());
    }
};
 
 
template<class MessageType, class QueueType>
class StateFlow : public TypedStateFlow<MessageType, UntypedStateFlow<QueueType> > {
public:
    StateFlow(Service *service)
        : TypedStateFlow<MessageType, UntypedStateFlow<QueueType>>(service)
    {
    }
};
 
#endif /* _EXECUTOR_STATEFLOW_HXX_ */