Dispatcher.hxx

Go to the documentation of this file.

 
#ifndef _EXECUTOR_DISPATCHER_HXX_
#define _EXECUTOR_DISPATCHER_HXX_
 
#include <vector>
 
#include "executor/Notifiable.hxx"
#include "executor/StateFlow.hxx"
 
template <int NUM_PRIO>
class DispatchFlowBase : public UntypedStateFlow<QList<NUM_PRIO>>
{
public:
    DispatchFlowBase(Service *service);
 
    typedef StateFlowBase::Action Action;
 
    ~DispatchFlowBase();
 
    size_t size();
 
protected:
    typedef uint32_t ID;
    typedef void UntypedHandler;
    void register_handler(UntypedHandler *handler, ID id, ID mask);
 
    void unregister_handler(UntypedHandler *handler, ID id, ID mask);
 
    void unregister_handler_all(UntypedHandler *handler);
 
    void register_fallback_handler(UntypedHandler *handler)
    {
        HASSERT(!fallbackHandler_);
        fallbackHandler_ = handler;
    }
 
    virtual ID get_message_id() = 0;
 
    virtual Action allocate_and_clone() = 0;
 
    virtual void send_transfer() = 0;
 
    /*typedef typename StateFlow<MessageType, QList<NUM_PRIO>>::Callback Callback;
    using StateFlow<MessageType, QList<NUM_PRIO>>::again;
    using StateFlow<MessageType, QList<NUM_PRIO>>::allocate_and_call;
    using StateFlow<MessageType, QList<NUM_PRIO>>::get_allocation_result;
    using StateFlow<MessageType, QList<NUM_PRIO>>::message;
    using StateFlow<MessageType, QList<NUM_PRIO>>::release_and_exit;
    using StateFlow<MessageType, QList<NUM_PRIO>>::transfer_message;*/
    using StateFlowBase::call_immediately;
    using StateFlowBase::again;
    using StateFlowWithQueue::release_and_exit;
 
    STATE_FLOW_STATE(entry) override;
 
    /*    Action entry()
        {
            currentIndex_ = 0;
            lastHandlerToCall_ = nullptr;
            return call_immediately(STATE(iterate));
            }*/
 
    STATE_FLOW_STATE(iterate);
    STATE_FLOW_STATE(clone_done);
    STATE_FLOW_STATE(iteration_done);
 
private:
    bool negateMatch_;
    template<class T>
    friend class GenericHubFlow;
 
    struct HandlerInfo
    {
        HandlerInfo() : handler(nullptr)
        {
        }
        ID id; 
        ID mask; 
        UntypedHandler *handler;
 
        bool Equals(ID id, ID mask, UntypedHandler *handler)
        {
            return (this->id == id && this->mask == mask &&
                    this->handler == handler);
        }
    };
 
    vector<HandlerInfo> handlers_;
 
    size_t currentIndex_;
 
protected:
    UntypedHandler *lastHandlerToCall_{nullptr};
    UntypedHandler *fallbackHandler_{nullptr};
private:
    OSMutex lock_;
};
 
 
// ==================== TEMPLATED =====================
 
#ifdef TARGET_LPC11Cxx
#define BASE_NUM_PRIO 4
#else
#define BASE_NUM_PRIO NUM_PRIO
#endif
 
template <class MessageType, int NUM_PRIO>
class DispatchFlow : public TypedStateFlow<MessageType, DispatchFlowBase<BASE_NUM_PRIO> > {
public:
    typedef TypedStateFlow<MessageType, DispatchFlowBase<BASE_NUM_PRIO> > Base;
 
    DispatchFlow(Service *service)
        :  Base(service) {}
 
    typedef StateFlowBase::Action Action;
    using StateFlowBase::call_immediately;
    using StateFlowBase::allocate_and_call;
    using StateFlowBase::get_allocation_result;
 
    typedef FlowInterface<MessageType> HandlerType;
    typedef typename MessageType::value_type::id_type ID;
 
    Action entry() OVERRIDE {
        return DispatchFlowBase<BASE_NUM_PRIO>::entry();
    }
 
    void register_handler(HandlerType *handler, ID id, ID mask) {
        Base::register_handler(handler, id, mask);
    }
 
    void unregister_handler(HandlerType *handler, ID id, ID mask) {
        Base::unregister_handler(handler, id, mask);
    }
 
    void unregister_handler_all(HandlerType *handler)
    {
        Base::unregister_handler_all(handler);
    }
 
    void register_fallback_handler(HandlerType *handler)
    {
        Base::register_fallback_handler(handler);
    }
 
protected:
    typename Base::ID get_message_id() OVERRIDE {
        return this->message()->data()->id();
    }
 
    Action allocate_and_clone() OVERRIDE {
        if (!this->lastHandlerToCall_) {
            // got unregistered.
            return call_immediately(STATE(clone_done));
        }
        HandlerType* h = static_cast<HandlerType *>(this->lastHandlerToCall_);
        return allocate_and_call(h, STATE(clone));
    }
 
    Action clone() {
        HandlerType* h = static_cast<HandlerType *>(this->lastHandlerToCall_);
        if (!this->lastHandlerToCall_) {  // got unregistered
            BufferBase* b;
            this->cast_allocation_result(&b);
            if (b) this->get_allocation_result(h)->unref();
            return call_immediately(STATE(clone_done));
        }
        MessageType *copy = this->get_allocation_result(h);
        copy->set_done(this->message()->new_child());
        *copy->data() = *this->message()->data();
        h->send(copy);
        return call_immediately(STATE(clone_done));
    }
 
    void send_transfer() OVERRIDE {
        HandlerType* h = static_cast<HandlerType *>(this->lastHandlerToCall_);
        h->send(this->transfer_message());
    }
};
 
 
// ================== IMPLEMENTATION ==================
 
template <int NUM_PRIO>
DispatchFlowBase<NUM_PRIO>::DispatchFlowBase(Service *service)
    : UntypedStateFlow<QList<NUM_PRIO>>(service)
    , negateMatch_(false)
    , lastHandlerToCall_(nullptr)
{
}
 
template<int NUM_PRIO>
DispatchFlowBase<NUM_PRIO>::~DispatchFlowBase()
{
    HASSERT(this->is_waiting());
}
 
template<int NUM_PRIO>
size_t DispatchFlowBase<NUM_PRIO>::size()
{
    OSMutexLock h(&lock_);
    size_t ret = 0;
    for (auto &h : handlers_)
    {
        if (h.handler)
        {
            ++ret;
        }
    }
    return ret;
}
 
template<int NUM_PRIO>
void DispatchFlowBase<NUM_PRIO>::register_handler(UntypedHandler *handler,
                                                  ID id, ID mask)
{
    OSMutexLock h(&lock_);
    size_t idx = 0;
    while (idx < handlers_.size() && handlers_[idx].handler)
    {
        ++idx;
    }
    if (idx >= handlers_.size())
    {
        handlers_.resize(handlers_.size() + 1);
    }
    handlers_[idx].handler = handler;
    handlers_[idx].id = id;
    handlers_[idx].mask = mask;
}
 
template<int NUM_PRIO>
void
DispatchFlowBase<NUM_PRIO>::unregister_handler(UntypedHandler *handler,
                                               ID id, ID mask)
{
    OSMutexLock h(&lock_);
    size_t idx = 0;
    while (idx < handlers_.size() && !handlers_[idx].Equals(id, mask, handler))
    {
        ++idx;
    }
    // Checks that we found the thing to unregister.
    HASSERT(idx < handlers_.size() &&
            "Tried to unregister a handler not previously registered.");
    if (lastHandlerToCall_ == handlers_[idx].handler) {
        lastHandlerToCall_ = nullptr;
    }
    handlers_[idx].handler = nullptr;
    if (idx == handlers_.size() - 1)
    {
        handlers_.resize(handlers_.size() - 1);
    }
}
 
template<int NUM_PRIO>
void DispatchFlowBase<NUM_PRIO>::unregister_handler_all(
    UntypedHandler *handler)
{
    OSMutexLock h(&lock_);
    for (size_t i = 0; i < handlers_.size(); ++i)
    {
        if (handlers_[i].handler == handler)
        {
            handlers_[i].handler = nullptr;
        }
    }
    while (!handlers_.empty() && handlers_.back().handler == nullptr)
    {
        handlers_.pop_back();
    }
}
 
template<int NUM_PRIO>
StateFlowBase::Action DispatchFlowBase<NUM_PRIO>::entry()
{
    currentIndex_ = 0;
    lastHandlerToCall_ = nullptr;
    return call_immediately(STATE(iterate));
}
 
template<int NUM_PRIO>
StateFlowBase::Action DispatchFlowBase<NUM_PRIO>::iterate()
{
    ID id = get_message_id();
    {
        // @todo(balazs.racz) make the registered handlers structure for the
        // dispatcher lock-free. This mutex here is very expensive.
        OSMutexLock l(&lock_);
        for (; currentIndex_ < handlers_.size(); ++currentIndex_)
        {
            auto &h = handlers_[currentIndex_];
            if (!h.handler)
            {
                continue;
            }
            if (negateMatch_ && (id & h.mask) == (h.id & h.mask))
            {
                continue;
            }
            if ((!negateMatch_) && (id & h.mask) != (h.id & h.mask))
            {
                continue;
            }
            // At this point: we have another handler.
            if (!lastHandlerToCall_)
            {
                // This was the first we found.
                lastHandlerToCall_ = handlers_[currentIndex_].handler;
                continue;
            }            
            break;
        }
    }
    if (currentIndex_ >= handlers_.size())
    {
        return iteration_done();
    }
    // Now: we have at least two different handler. We need to clone the
    // message. We use the pool of the last handler to call by default.
    return allocate_and_clone();
}
 
template<int NUM_PRIO>
StateFlowBase::Action DispatchFlowBase<NUM_PRIO>::clone_done()
{
    lastHandlerToCall_ = handlers_[currentIndex_].handler;
    ++currentIndex_;
    return call_immediately(STATE(iterate));
}
 
template<int NUM_PRIO>
StateFlowBase::Action DispatchFlowBase<NUM_PRIO>::iteration_done()
{
    if (lastHandlerToCall_)
    {
        send_transfer();
    }
    else if (fallbackHandler_)
    {
        // Nothing handled this message, and we have a fallbac handler
        // registered. Gives the message to the fallback handler.
        lastHandlerToCall_ = fallbackHandler_;
        send_transfer();
    }
    return release_and_exit();
}
 
#endif // _EXECUTOR_DISPATCHER_HXX_