root/lang/objective-cplusplus/i3/trunk/src/mil/include/mil/Module.h @ 32286

#pragma once
#include "Mil.h"
#include "Thread.h"
#include "ModuleExecuteProxy.h"

namespace mil
{

/**
 * public members should be thread safe
 */
template <typename Child>
class Module : public Thread<Module<Child> >
{
public:
    pool::Producer producer;
    pool::Consumer consumer;
private:
    class EventBase : public boost::intrusive::slist_base_hook<>, public boost::noncopyable
    {
    public:
        void (*execute)(void* event, void* target);
        intptr_t owner_id;
    };

    class EventList : public boost::noncopyable
    {
        boost::intrusive::slist<EventBase, boost::intrusive::constant_time_size<false>, boost::intrusive::cache_last<true> > l;
        boost::details::pool::default_mutex m;
        typedef boost::details::pool::guard<boost::details::pool::default_mutex> guard;
    public:
        void push(EventBase& e)
        {
            guard g(m);
            l.push_back(e);
        }

        void pop()
        {
            guard g(m);
            l.pop_front();
        }

        EventBase* top()
        {
            guard g(m);
            if (l.size() == 0)
            {
                return NULL;
            }
            return &l.front();
        }
    };

    template <typename T>
    class Event : public EventBase
    {
    public:
        T data;
    };

    friend class Module_test;

private:
    EventList list;
    Conditional c;
    bool repost;
    atomic<bool> breakLoopRequest;
    atomic<bool> isWaiting;

public:
    Module() : repost(false), breakLoopRequest(false), isWaiting(false)
    {
    }

    Child& getChild()
    {
        return *static_cast<Child*>(this);
    }

protected:
    void repostEvent()
    {
        repost = true;
    }


    template <typename Data, typename Target>
    static void dispatcher(void* event_, void* target_)
    {
        if (!event_ || !target_)
        {
            return;
        }

        // XXX type safety??
        Target& target = *static_cast<Target*>(target_);
        Event<Data>& event = *static_cast<Event<Data>*>(event_);
        target.list.pop();
        target.repost = false;
        ModuleExecuteProxy<Target>::execute(target, event.data);
        if (target.repost) {
            target.list.push(event);
            return;
        }
        pool::MemoryList* m = target.consumer.work((void*)&event, sizeof(event), event.owner_id);
        if (!m) {
            return;
        }
        pool::ReturnMemoryEvent e;
        e.memory = m;
        void* t = threads[event.owner_id].load();
        if (!t) {
            return;
        }
        (reinterpret_cast<Module<Child> * >(t))->post(e, target.producer);
    }
public:
    template <typename T>
    void post(const T& event, mil::pool::Producer& sender_producer)
    {
        void* p = sender_producer.malloc<sizeof(Event<T>)>();
        if (p == NULL)
        {
            // XXX
            return;
        }

        Event<T>* e = new(p) Event<T>;
        e->data = event;
        e->execute = dispatcher<T, Child>;
        e->owner_id = this->thread_id;

        list.push(*e);
        if (isWaiting.load()) {
            c.signal();
        }
    }
public:
    void destroy()
    {
        breakLoopRequest.store(true);
    }
    ~Module() {
        destroy();
    }
    void execute_front()
    {
        EventBase* e = list.top();
        if (e)
        {
            e->execute((void*)e, (void*)this);
        }
    }

protected:
    void loop()
    {
        while (!breakLoopRequest.load())
        {
            EventBase* e = list.top();
            if (e)
            {
                e->execute((void*)e, (void*)this);
            }
            else
            {
                isWaiting.store(true);
                c.wait();
                isWaiting.store(false);
            }
        }
        breakLoopRequest.store(false);
    }
};

}