src/actor.cpp

/*                                                                            *\
**                          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/>.             **
\*                                                                            */

#define ACEDIA_REDUCED_ANNOUNCE

#ifndef _XOPEN_SOURCE
#  define _XOPEN_SOURCE
#endif

#include "acedia_config.hpp"
#include "actor.hpp"
#include "actorref.hpp"
#include "exceptions.hpp"

#include <boost/bind.hpp>

#include <iostream>

typedef boost::lock_guard<boost::mutex> MutexLocker;

// NOTE: implementation of doExit() is in AbstractActor.cpp
namespace acedia
{

    static const int DETACHED_OR_HIDDEN = DETACHED | HIDDEN;

    void Actor::reply(const Any &val1)
    {
        send(m_lrm.sender(), val1);
    }

    void Actor::reply(const acedia::Any &val1, const acedia::Any &val2)
    {
        send(m_lrm.sender(),val1,val2);
    }

    struct NopProcessor : public details::AnyArrayProcessor
    {
        details::ContinuationInvoker ci;

        NopProcessor() : ci(this, NULL) { }

        virtual bool operator()(const AnyArray &) { return true; }

        virtual details::ContinuationInvoker *matchingInvoker(const AnyArray &)
        {
            return &ci;
        }
    };

    bool Actor::tryReceive(Message &storage)
    {
        details::container::SingleReaderListIterator<Message> i(m_mailbox);
        if (i.hasNext())
        {
            m_lrm = i.next();
            storage = m_lrm;
            i.erase();
            yield(READY);
            return true;
        }
        yield(READY);
        return false;
    }

    Message Actor::receive()
    {
        Invoker i;
        i.add(new NopProcessor);
        receiveAndInvoke(i);
        ACEDIA_MEMORY_BARRIER();
        return m_lrm;
    }

    bool Actor::tryReceiveAndInvoke(Invoker &imp)
    {
        details::ContinuationInvoker *invoker;
        details::container::SingleReaderListIterator<Message> i(m_mailbox);
        while (i.hasNext())
        {
            Message msg = i.next();
            invoker = imp.matchingInvokerTo(msg, m_trapExit);
            if (invoker)
            {
                m_lrm = msg;
                i.erase();
                yield(READY);
                ACEDIA_MEMORY_BARRIER();
                (*invoker)();
                return true;
            }
        }
        yield(READY);
        return false;
    }

    void setBoolToFalse(bool* b)
    {
        *b = false;
    }

    void doNothingCallback() { }

    bool Actor::receiveWithin(Message& storage, boost::uint16_t msTimeout)
    {
        Invoker nopInvoker;
        nopInvoker.add(others() >> boost::function0<void>(doNothingCallback));
        if (receiveAndInvokeWithin(nopInvoker, msTimeout))
        {
            storage = lastReceivedMessage();
            return true;
        }
        return false;
    }

    void Actor::receiveAndInvoke(Invoker &imp)
    {
        bool success = false;
        {
            details::ContinuationInvoker *invoker;
            details::container::SingleReaderListIterator<Message> i(m_mailbox);
            while (!success && i.hasNext())
            {
                Message msg = i.next();
                invoker = imp.matchingInvokerTo(msg, m_trapExit);
                if (invoker)
                {
                    m_lrm = msg;
                    i.erase();
                    success = true;
                    m_continuationInvoker = invoker;
                }
            }
            if (!success)
            {
                // wait for incoming messages: get exclusive mailbox access
                util::ExclusiveLockGuard guard(m_mailboxLock);
                // exclusive access to the list
                // try again to read a matching element
                while (!success && i.hasNext())
                {
                    Message msg = i.next();
                    invoker = imp.matchingInvokerTo(msg, m_trapExit);
                    if (invoker)
                    {
                        m_lrm = msg;
                        i.erase();
                        success = true;
                        m_continuationInvoker = invoker;
                    }
                }
                if (!success)
                {
                    // nothing found
                    m_continuationInvoker = NULL;
                    m_receptionist = &imp;
                }
            }
        }
        m_lrmSet = success;
        ACEDIA_MEMORY_BARRIER();
        // told the scheduler what our next state is
        if (success)
        {
            // we could continue
            yield(READY);
        }
        else
        {
            // yield blocked until m_continuationInvoker is != NULL
            // the loop is needed because detached actors may be resumed
            // before lrm was set
            do
            {
                yield(BLOCKED);
            }
            while (NULL == m_continuationInvoker);
        }
        ACEDIA_MEMORY_BARRIER();
        // continuation point. Invoker etc are definitely set now
        (*const_cast<details::ContinuationInvoker*>(m_continuationInvoker))();
        m_continuationInvoker = NULL;
    }

    bool Actor::setState(ActorState oldState, ActorState nextState)
    {
        if (nextState == EXITED)
        {
            bool result = false;
            if (atomic::compareAndSwap(&m_state, oldState, EXITED))
            {
                {
                    // get exclusive mailbox access and clear it
                    util::ExclusiveLockGuard guard(m_mailboxLock);
                    // ok, we have exclusive access now
                    m_lrm = Message();
                    // clear mailbox now
                    m_mailbox.clear();
                }
                result = true;
                // send Exit messages to all links
                {
                    MutexLocker locker(m_linksMutex);
                    Any exitTuple = Exit(m_exitReason);
                    for (std::list<ActorRef>::iterator i = m_links.begin(); i != m_links.end(); ++i)
                    {
                        (*i).send(Message(self(), *i, exitTuple));
                    }
                    m_links.clear();
                }
                ACEDIA_MEMORY_BARRIER();
                // run onExit (and ignore exceptions)
                try { onExit(); }
                catch (...) { }
            }
            return result;
        }
        else
        {
            return atomic::compareAndSwap(&m_state, oldState, nextState);
        }
    }


    void Actor::resume(Context *schedulerCtx)
    {
        // schedulerContext won't be changed outside this method and this method
        // is called exactly by one thread at a time.
        m_returnContext = const_cast<volatile Context*>(schedulerCtx);
        ACEDIA_MEMORY_BARRIER();
        Context::swap(*schedulerCtx, *m_context);
    }

    void Actor::setLrm(const Message &msg)
    {
        m_lrm = msg;
        m_receptionist = NULL;
        m_timeoutReceptionist = NULL;
        // if the actor is still running wait until his new state was set
        ActorState as = m_state;
        if (as == RUNNING)
        {
            while ((as = m_state) & RUNNING) ; // nop
        }
        // loop to modify the state
        // this loop is needed because a blocked/waiting actor could be
        // detached and the first "try" to call setState may fail
        // it's also possible that the actor is still marked as RUNNING
        // we have to wait in this case also
        for (;;)
        {
            assert(as & (BLOCKED | WAITING | DETACHED_OR_HIDDEN));
            if (as & (BLOCKED | WAITING) && setState(as, READY))
            {
                // release locks
                m_lrmSet = true;
                // state is now READY
                // -> enqueue to the scheduler
                ACEDIA_MEMORY_BARRIER();
                details::enqueueToScheduler(this);
                return;
            }
            else if (as & DETACHED_OR_HIDDEN)
            {
                // the actor is detached
                // do only release locks
                // release locks
                m_lrmSet = true;
                ACEDIA_MEMORY_BARRIER();
                // don't enqueue to the scheduler, just notify
                // the waiting worker
                m_lrmSetCondition.notify_one();
                return;
            }
            else as = m_state; // next loop value
        }
    }

    bool Actor::receiveAndInvokeWithin(Invoker &imp, boost::uint16_t msTimeout)
    {
        bool msgReceived = true;
        // request timeout
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        // 0 is never used as request id
        while (0 == ++m_timeoutRequestId) { }
        boost::int32_t reqId = m_timeoutRequestId;
        Invoker toutInvoker(on<TimeoutEvent>().unbox().guard(isEq(reqId))
                            >> boost::function0<void>(boost::bind(setBoolToFalse, &msgReceived)));
        m_timeoutReceptionist = &toutInvoker;
        send(timeEmitter, TimeoutRequest(reqId, msTimeout));
        // receive message (or timeout)
        receiveAndInvoke(imp);
        return msgReceived;
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter, FutureMessageRequest(msTimeout), v1);
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1, const Any& v2)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2);
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1, const Any& v2, const Any& v3)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3);
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1, const Any& v2, const Any& v3, const Any& v4)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3,v4);
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1, const Any& v2, const Any& v3, const Any& v4, const Any& v5)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3,v4,v5);
    }

    void Actor::futureMessage(boost::uint32_t msTimeout, const Any& v1, const Any& v2, const Any& v3, const Any& v4, const Any& v5, const Any& v6)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3,v4,v5,v6);
    }

    void Actor::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)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3,v4,v5,v6,v7);
    }

    void Actor::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)
    {
        ActorRef timeEmitter = whereisName("acedia_TimeEmitter");
        send(timeEmitter,FutureMessageRequest(msTimeout),v1,v2,v3,v4,v5,v6,v7,v8);
    }

    bool Actor::filterTimeoutEvents(const Message& msg)
    {
        if (msg.match<TimeoutEvent>())
        {
            // lock mailbox for write (shared lock)
            util::SharedLockGuard mboxGuard(m_mailboxLock);
            // check condition again (with the lock)
            if (state() != EXITED)
            {
                if (m_timeoutReceptionist)
                {
                    // there is a receptionist waiting for us
                    util::SpinlockGuard recGuard(m_receptionistLock);
                    Invoker* r = const_cast<Invoker*>(m_timeoutReceptionist);
                    if (r) // check condition again (with lock)
                    {
                        details::ContinuationInvoker *i =
                                r->matchingInvokerTo(msg, m_trapExit);
                        if (i)
                        {
                            m_continuationInvoker = i;
                            Message lrmAgain = lastReceivedMessage();
                            setLrm(lrmAgain);
                        }
                    }
                }
            }
            return true;
        }
        return false;
    }

    bool Actor::filterIncomingMessage(const Message& msg)
    {
        if (msg.match<AddRemoteLink>())
        {
            // do only add the sender to the list and don't backlink
            ActorRef who = msg.sender();
            if (who != self())
            {
                MutexLocker locker(m_linksMutex);
                if (m_state == EXITED) send(who, Exit(exitReason()));
                else m_links.push_back(who);
            }
            return true;
        }
        else return filterTimeoutEvents(msg);
    }

    void Actor::enqueue(const Message &msg)
    {
        if (!filterIncomingMessage(msg) && state() != EXITED)
        {
            // lock mailbox for write (shared lock)
            util::SharedLockGuard mboxGuard(m_mailboxLock);
            // check condition again (with the lock)
            if (state() != EXITED)
            {
                if (m_receptionist)
                {
                    // there is a receptionist waiting for us
                    util::SpinlockGuard recGuard(m_receptionistLock);
                    Invoker *r = const_cast<Invoker*>(m_receptionist);
                    if (r) // check condition again (with lock)
                    {
                        details::ContinuationInvoker *i =
                                r->matchingInvokerTo(msg, m_trapExit);
                        if (i)
                        {
                            m_continuationInvoker = i;
                            setLrm(msg);
                        }
                        else
                        {
                            m_mailbox.append(msg);
                        }
                    }
                    else // someone else set the receptionist to NULL
                    {
                        m_mailbox.append(msg);
                    }
                }
                else // no receptionist is waiting for us
                {
                    // just enqueue the message and leave
                    m_mailbox.append(msg);
                }
            }
        }

/*
        // loop to get write access to the mailbox
        if (!filterIncomingMessage(msg) && state() != EXITED) for (;;)
        {
            boost::int32_t v = m_mailboxLock;
            // write access is granted if the spin lock has a value >= 0
            // and the CAS operation returns true
            if (v >= 0 && atomic::compareAndSwap(&m_mailboxLock, v, v+1))
            {
                // check condition again
                if (state() == EXITED)
                {
                    (void) atomic::addAndFetch(&m_mailboxLock, -1);
                    return;
                }
                // ok, it's safe to write now
                if (m_receptionist)
                {
                    // there is a receptionist waiting for us
                    for (;;) // loop to lock receptionistSpinLock
                    {
                        if (atomic::compareAndSwap(
                                &m_receptionistLock, 0, 1))
                        {
                            Invoker *r = const_cast<Invoker*>(m_receptionist);
                            if (r) // check condition again
                            {
                                details::ContinuationInvoker *i =
                                        r->matchingInvokerTo(msg, m_trapExit);
                                if (i)
                                {
                                    m_continuationInvoker = i;
                                    setLrm(msg);
                                    return;
                                }
                                else
                                {
                                    enqueueImpl(msg, true);
                                    return;
                                }
                            }
                            else
                            {
                                // someone else set the receptionist to NULL
                                enqueueImpl(msg, true);
                                return;
                            }
                        }
                    }
                }
                // no receptionist is waiting for us
                // just enqueue the message and leave
                else
                {
                    enqueueImpl(msg);
                    return;
                }
            }
        }
*/
    }

    void Actor::yield(ActorState mYieldState)
    {
        if (mYieldState == EXITED) throw ActorKilledException(m_exitReason);
        else if (isDetached() && mYieldState == READY)
        {
            // don't return to the scheduler-worker because he would
            // return immediately
            return;
        }
        *yieldState = mYieldState;
        ACEDIA_MEMORY_BARRIER();
        Context::swap(*m_context, *const_cast<Context*>(m_returnContext));
        ACEDIA_MEMORY_BARRIER();
    }

    Actor::Actor(ActorState initialState, bool tExit) :
            AbstractActor(false, false),
            m_state(initialState), m_context(NULL),
            m_returnContext(NULL),
            m_receptionist(NULL), m_timeoutReceptionist(NULL),
            m_trapExit(tExit),
            m_lrmSet(false), m_timeoutRequestId(0),
            next(NULL), yieldState(NULL)
    {
        assert(initialState & (UNINITIALIZED | DETACHED_OR_HIDDEN));
        if (initialState & DETACHED_OR_HIDDEN)
        {
            m_state = UNINITIALIZED | initialState;
        }
    }

    void Actor::initialize()
    {
#if defined(ACEDIA_UCONTEXT_IMPL)
        m_context = new Context((void (*)()) &Actor::ucontext_trampoline,
                                (int) this);
#elif defined(ACEDIA_FIBER_IMPL)
        m_context = new Context((LPFIBER_START_ROUTINE) &Actor::fiber_trampoline,
                                (LPVOID) this);
#endif
        if (m_state == UNINITIALIZED)
        {
            assert(setState(UNINITIALIZED, READY));
        }
        else if (m_state == (UNINITIALIZED | DETACHED))
        {
            assert(setState((UNINITIALIZED | DETACHED), DETACHED));
        }
        else if (m_state == (UNINITIALIZED | HIDDEN))
        {
            assert(setState((UNINITIALIZED | HIDDEN), HIDDEN));
        }
        else
        {
            assert(false);
        }
    }

    Actor::~Actor()
    {
        if (m_context) delete m_context;
        // ok, delete mailbox now
        m_mailbox.clear();
    }

    void Actor::linkTo(const ActorRef &who)
    {
        if (who == self()) return;
        MutexLocker locker(m_linksMutex);
        ActorRef s = self();
        if (m_state == EXITED)
        {
            ActorRef receiver = who;
            send(receiver, Exit(exitReason()));
        }
        else if (who.actor->backlinkTo(s))
        {
            m_links.push_back(who);
        }
    }

    // add who to the link-list and return true
    bool Actor::backlinkTo(ActorRef &who)
    {
        if (who == self()) return false;
        {
            MutexLocker locker(m_linksMutex);
            if (m_state == EXITED)
            {
                send(who, Exit(exitReason()));
            }
            else m_links.push_back(who);
        }
        return true;
    }

    void Actor::onExit() { }

    void Actor::doExit(boost::int32_t reason)
    {
        throw ActorKilledException(reason);
    }

    void Actor::normalExit()
    {
        doExit(exit_reasons::NORMAL_EXIT);
    }

    void appStr(String &str, const char *what)
    {
        if (str.empty()) str += what;
        else
        {
            str += "|";
            str += what;
        }
    }

    String Actor::stateAsString() const
    {
        String str;
        ActorState s = m_state;
        if (s & UNINITIALIZED) appStr(str, "UNINITIALIZED");
        if (s & READY) appStr(str, "READY");
        if (s & BLOCKED) appStr(str, "BLOCKED");
        if (s & WAITING) appStr(str, "WAITING");
        if (s & RUNNING) appStr(str, "RUNNING");
        if (s & DETACHED) appStr(str, "DETACHED");
        if (s & HIDDEN) appStr(str, "HIDDEN");
        if (s & EXITED) appStr(str, "EXITED");
        return str;
    }

} // namespace acedia

---