acedia/actor.hpp

/*                                                                            *\
**                          Copyright (C) 2009-2010                           **
**       Dominik Charousset < Dominik.Charousset (at) haw-hamburg.de >        **
**                                                                            **
**            MMM       MMMMMM   MMMMMMMMM MMMMMMM   MMM       MMM            **
**           MMMMM     MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM      MMMMM           **
**          MMM MMM   MMM        MMMM      MMM   MMM MMM     MMM MMM          **
**         MMM   MMM  MMM        MMMMMMMMM MMM   MMM MMM    MMM   MMM         **
**        MMMMMMMMMMM MMMM   MMM MMMM      MMM   MMM MMM   MMMMMMMMMMM        **
**       MMM       MMM MMMMMMMM  MMMMMMMMM MMMMMMMM  MMM  MMM       MMM       **
**      MMM         MMM MMMMMM   MMMMMMMMM MMMMMMM   MMM MMM         MMM      **
**                                                                            **
**                                                                            **
** This file is part of the acedia library.                                   **
**                                                                            **
** This library is free software: you can redistribute it and/or modify       **
** it under the terms of the GNU Lesser General Public License as published   **
** by the Free Software Foundation, either version 3 of the License, or       **
** (at your option) any later version.                                        **
**                                                                            **
** This library is distributed in the hope that it will be useful,            **
** but WITHOUT ANY WARRANTY; without even the implied warranty of             **
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the               **
** GNU Lesser General Public License for more details.                        **
**                                                                            **
** You should have received a copy of the GNU Lesser General Public License   **
** along with acedia. If not, see <http://www.gnu.org/licenses/>.             **
\*                                                                            */

#ifndef ACEDIA_ACTOR_H_
#define ACEDIA_ACTOR_H_

#ifndef _XOPEN_SOURCE
#  define _XOPEN_SOURCE
#endif

#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/static_assert.hpp>
#include <boost/thread/condition_variable.hpp>

#include "any.hpp"
#include "acedia_atomic.hpp"
#include "acedia.hpp"
#include "context.hpp"
#include "invoker.hpp"
#include "actorref.hpp"
#include "abstractactor.hpp"
#include "tuple.hpp"

#include "acedia_util.hpp"
#include "acedia_details.hpp"

namespace acedia
{

    typedef boost::int32_t ActorState;

    static const boost::int32_t UNINITIALIZED = 0x01;

    static const boost::int32_t READY = 0x02;

    static const boost::int32_t BLOCKED = 0x04;

    static const boost::int32_t WAITING = 0x08;

    static const boost::int32_t RUNNING = 0x10;

    static const boost::int32_t DETACHED = 0x20;

    static const boost::int32_t HIDDEN = 0x40;


    static const boost::int32_t EXITED = 0x80;

    class Actor : public AbstractActor
    {
        friend class ActorRef;
        friend ActorRef details::doSpawn(Actor*);
        friend struct Worker;

        static void trampoline(Actor *_this);

#if defined(ACEDIA_UCONTEXT_IMPL)
        // if we are forced to cast (Actor*) to int let's at least ensure
        // that it's safe
        BOOST_STATIC_ASSERT(sizeof(Actor*) == sizeof(int));
        // static helper method to "jump" into
        // an object scope after makecontext
        static void ucontext_trampoline(int _this);
#elif defined(ACEDIA_FIBER_IMPL)
        static void fiber_trampoline(PVOID *_this);
#endif

        // 'main' method of an actor
        void executeActor();

        // the actors state
        volatile ActorState m_state;
        // the actors context
        Context *m_context;
        // the context to which yield() should swap back
        volatile Context *m_returnContext;

        // used to synchronize access to the mailbox
        util::RWSpinlock m_mailboxLock;
        // protects both m_receptionist and m_timeoutReceptionist
        util::Spinlock m_receptionistLock;
        // the mailbox
        details::container::SingleReaderList<Message> m_mailbox;

        // the receptionist (if set) checks incoming mails against the current
        // active patterns and sets continuationInvoker on a matching message
        volatile Invoker* m_receptionist;
        volatile Invoker* m_timeoutReceptionist;
        // if an actor yields because he waits for a new message to proccess
        // then he will call this invoker after resume
        volatile acedia::details::ContinuationInvoker *m_continuationInvoker;
        // lrm = Last Received Message
        Message m_lrm;

        bool m_trapExit;

        // links are stored in a list ...
        std::list<ActorRef> m_links;
        // ... which is guarded by a mutex
        boost::mutex m_linksMutex;

        // needed by detached actors to sync
        volatile bool m_lrmSet;
        boost::condition_variable m_lrmSetCondition;

        void setLrm(const Message &msg);

        // counter to produce one-time ids
        boost::int32_t m_timeoutRequestId;

    protected:

        bool filterTimeoutEvents(const Message &msg);

        virtual bool filterIncomingMessage(const Message &msg);

        void doExit(boost::int32_t reason);

        void normalExit();

        inline Message lastReceivedMessage() const { return m_lrm; }

        inline void forwardLastReceivedMessage(ActorRef &whom) const
        {
            whom.send(m_lrm);
        }

        virtual void act() = 0;

        virtual void onExit();

        void yield(ActorState mYieldState = READY);

        void receiveAndInvoke(Invoker &imp);

        template<class CaseClass>
        void waitFor()
        {
            Invoker inv;
            inv.add(on<CaseClass>() >> discard);
            receiveAndInvoke(inv);
        }

        bool tryReceiveAndInvoke(Invoker &imp);

        bool receiveAndInvokeWithin(Invoker &imp, boost::uint16_t msTimeout);

        Message receive();

        bool tryReceive(Message &storage);

        bool receiveWithin(Message &storage, boost::uint16_t msTimeout);

        void reply(const Any &val1);

        void reply(const acedia::Any &val1, const acedia::Any &val2);

        template<class A>
        ActorRef spawn_link()
        {
            ActorRef actor = spawn<A>();
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1>
        ActorRef spawn_link(const T1 &v1)
        {
            ActorRef actor = spawn<A>(v1);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2>
        ActorRef spawn_link(const T1 &v1, const T2 &v2)
        {
            ActorRef actor = spawn<A>(v1,v2);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3)
        {
            ActorRef actor = spawn<A>(v1,v2,v3);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4,
                typename T5>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4, const T2 &v5)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4,v5);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4,
                typename T5, typename T6>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4, const T2 &v5, const T3 &v6)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4,v5,v6);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4,
                typename T5, typename T6, typename T7>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4, const T2 &v5, const T3 &v6,
                            const T1 &v7)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4,v5,v6,v7);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4,
                typename T5, typename T6, typename T7, typename T8>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4, const T2 &v5, const T3 &v6,
                            const T1 &v7, const T2 &v8)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4,v5,v6,v7,v8);
            linkTo(actor);
            return actor;
        }

        template<class A, typename T1, typename T2, typename T3, typename T4,
                typename T5, typename T6, typename T7, typename T8, typename T9>
        ActorRef spawn_link(const T1 &v1, const T2 &v2, const T3 &v3,
                            const T1 &v4, const T2 &v5, const T3 &v6,
                            const T1 &v7, const T2 &v8, const T3 &v9)
        {
            ActorRef actor = spawn<A>(v1,v2,v3,v4,v5,v6,v7,v8,v9);
            linkTo(actor);
            return actor;
        }

        void futureMessage(boost::uint32_t msTimeout, const Any& v1);

        void futureMessage(boost::uint32_t msTimeout,
                           const Any& v1, const Any& v2);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3, const Any& v4);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3, const Any& v4,
                           const Any& v5);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3, const Any& v4,
                           const Any& v5, const Any& v6);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3, const Any& v4,
                           const Any& v5, const Any& v6, const Any& v7);

        void futureMessage(boost::uint32_t msTimeout, const Any& v1,
                           const Any& v2, const Any& v3, const Any& v4,
                           const Any& v5, const Any& v6, const Any& v7,
                           const Any& v8);

    public:

        Actor *next; // the job queue of the scheduler is intrusive

        inline const volatile ActorState &state() const { return m_state; }

        String stateAsString() const;

        /*
          the reference for yieldState is set by the scheduler before resume()
          is called and after yield() is called this value tolds the
          scheduler what the next state of this actor is
          (the scheduler afterwards calls setState(yieldState) if
          yieldState != EXITED
        */
        volatile ActorState *yieldState;

        bool setState(ActorState oldState, ActorState nextState);

        Actor(ActorState initialState = UNINITIALIZED, bool trapExit = false);

        void initialize();

        virtual ~Actor();

        void resume(Context *schedulerCtx);

        inline bool isBlocked() const { return m_state == BLOCKED; }

        inline bool isDetached() const
        {
            return 0 != (m_state & (DETACHED | HIDDEN));
        }

        /* override */ virtual bool backlinkTo(ActorRef &who);

        /* override */ virtual void linkTo(const ActorRef &who);

        /* override */ virtual void enqueue(const Message &msg);

    };

} // namespace acedia

namespace acedia
{
    struct Exit;

    struct AddRemoteLink;
}

ACEDIA_DECLARE_CASE_TUPLE1((acedia)(Exit), reason, boost::int32_t)

ACEDIA_DECLARE_CASE_CLASS((acedia)(AddRemoteLink))

// request a timeout
ACEDIA_DECLARE_CASE_TUPLE2((acedia)(TimeoutRequest), requestId, boost::int32_t, msTimeout, boost::uint32_t)

// the answer to a TimeoutRequest which refers to the requestId
ACEDIA_DECLARE_CASE_TUPLE1((acedia)(TimeoutEvent), requestId, boost::int32_t)

ACEDIA_DECLARE_CASE_TUPLE1((acedia)(FutureMessageRequest), msTimeout, boost::uint32_t)

#endif // ACEDIA_ACTOR_H_

---