src/acedia.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

// std includes
#include <list>
#include <sstream>
#include <set>

// boost includes
#include <boost/thread.hpp>
#include <boost/cstdint.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition_variable.hpp>

// acedia includes
#include "actor.hpp"
#include "acedia.hpp"
#include "acedia_atomic.hpp"
#include "acedia_util.hpp"

#include <iostream>
using std::cout;
using std::endl;

namespace acedia
{
    namespace details
    {
        class InvokeVoidFunctor : public AnyArrayProcessor
        {

            boost::function0<void> f;

        public:

            InvokeVoidFunctor(const boost::function0<void> &func) : f(func) { }

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

            virtual ContinuationInvoker *matchingInvoker(const AnyArray &arr)
            {
                return new ContinuationInvoker(this, &arr);
            }

        };

    }

    void discard() { }

    details::AnyArrayProcessor* MatchAllRuleBuilder::operator>>(void (*voidFunction)())
    {
        return ((*this) >> boost::function0<void>(voidFunction));
    }

    details::AnyArrayProcessor* MatchAllRuleBuilder::operator>>(const boost::function0<void> &f)
    {
        return new details::InvokeVoidFunctor(f);
    }

    static boost::int32_t schedulerStartetFlag = 0;

    static boost::int32_t runningActors = 0;

    static boost::condition_variable noRunningActorsCv;

    inline void incRunningActors()
    {
        (void) atomic::addAndFetch(&runningActors, 1);
    }

    inline void decRunningActors()
    {
        if (0 == atomic::addAndFetch(&runningActors, -1))
            noRunningActorsCv.notify_all();
    }

    /**************************************************************************
     *                              TimeEmitter                               *
     **************************************************************************/

    struct TimeEmitter : public Actor
    {

//      typedef std::pair<ActorRef, boost::int32_t> TMapPair;
        typedef std::multimap<boost::system_time /* timeout */,
                              Message /* message to send if timeout occurs */>
                TimeoutMap;
        typedef TimeoutMap::value_type TimeoutMapEntry;

        TimeoutMap timeouts;

        TimeEmitter() : Actor(HIDDEN), timeouts() { }

        void handleRequest(ActorRef requester, boost::int32_t reqId, boost::uint32_t msTimeout)
        {
            assert(requester.isValid());
            boost::system_time tOut = boost::get_system_time();
            tOut += boost::posix_time::milliseconds(msTimeout);
            timeouts.insert(TimeoutMapEntry(tOut,
                                            Message(requester, requester,
                                                    TimeoutEvent(reqId))));
//                                          std::make_pair(requester, reqId)));
        }

        bool timeoutHappens()
        {
            boost::system_time st = boost::get_system_time();
            TimeoutMap::iterator it = timeouts.begin();
            if (it != timeouts.end() && (it->first) < st)
            {
//              cout << __PRETTY_FUNCTION__ << endl;
//              TMapPair &tReq = it->second;
                Message delivery = it->second;
                ActorRef whom = delivery.receiver();
                whom.send(delivery);
//              send(tReq.first, TimeoutEvent(tReq.second));
                timeouts.erase(it);
                return true;
            }
            return false;
        }

        Message createFutureMessage(const Message& from)
        {
            ActorRef s = from.sender();
            // first element is the FutureMessageRequest
            // maximum of elements is 9 (given by the message constructor)
            switch (from.length())
            {
             case  2: return Message(s,s,from.at(1));
             case  3: return Message(s,s,from.at(1),from.at(2));
             case  4: return Message(s,s,from.at(1),from.at(2),from.at(3));
             case  5: return Message(s,s,from.at(1),from.at(2),from.at(3),from.at(4));
             case  6: return Message(s,s,from.at(1),from.at(2),from.at(3),from.at(4),from.at(5));
             case  7: return Message(s,s,from.at(1),from.at(2),from.at(3),from.at(4),from.at(5),from.at(6));
             case  8: return Message(s,s,from.at(1),from.at(2),from.at(3),from.at(4),from.at(5),from.at(6),from.at(7));
             case  9: return Message(s,s,from.at(1),from.at(2),from.at(3),from.at(4),from.at(5),from.at(6),from.at(7),from.at(8));
             default: return Message(s,s);
            }
        }

        virtual void act()
        {
            Message msg;
            MetaClass* timeoutRequestMetaClass =
                    getMetaClass("acedia::TimeoutRequest");
            MetaClass* futureMessageRequestMetaClass =
                    getMetaClass("acedia::FutureMessageRequest");
            assert(NULL != timeoutRequestMetaClass);
            for (;;)
            {
                while (tryReceive(msg))
                {
                    if (msg.metaClassAt(0) == timeoutRequestMetaClass)
                    {
                        const TimeoutRequest& tr =
                                msg.uncheckedValueAt<TimeoutRequest>(0);
                        handleRequest(msg.sender(), tr.requestId(), tr.msTimeout());
                    }
                    else if (msg.metaClassAt(0) == futureMessageRequestMetaClass)
                    {
                        if (msg.sender().isValid())
                        {
                            const FutureMessageRequest& req =
                                    msg.uncheckedValueAt<FutureMessageRequest>(0);
                            boost::uint32_t msTimeout = req.msTimeout();
                            boost::system_time tOut = boost::get_system_time();
                            tOut += boost::posix_time::milliseconds(msTimeout);
                            timeouts.insert(TimeoutMapEntry(tOut,
                                                            createFutureMessage(msg)));
                        }
                    }
                }
                // check for timeouts
                while (timeoutHappens())
                {
                    // nop
                }
                // sleep at least 1ms
                boost::system_time st = boost::get_system_time();
                st += boost::posix_time::milliseconds(1);
                boost::this_thread::sleep(st);
            }
        }

    };

    /**************************************************************************
     *                               Scheduler                                *
     **************************************************************************/

    // forward declaration
    struct Worker;

    struct SchedulerData
    {
        Worker *worker;
        boost::uint32_t lastReadedTStamp;
        boost::system_time timeout;
        inline SchedulerData() throw() :
                worker(NULL), lastReadedTStamp(UInt32Maximum),
                timeout(boost::get_system_time())
        {
        }
        ~SchedulerData();
    };

    struct Scheduler
    {

        volatile bool run;
        volatile bool exited;

        details::container::intrusive::SingleReaderQueue<Actor> jobQ;
        util::Semaphore jobsQSem;
        util::Spinlock jobQLock;

        Scheduler() : run(true), exited(false) { }

        // called by the workers; blocking call
        Actor *getJob()
        {
            // wait for at least one job available in the jobQ
            if (!run) return NULL;
            while (!jobsQSem.tryAcquire(1, 30))
            {
                if (!run) return NULL;
            }
            // fetch a job
            {
                util::SpinlockGuard guard(jobQLock);
                return jobQ.tryDequeue();
            }
        }

        // called by the workers; non-blocking call
        Actor *tryGetJob()
        {
//          cout << std::string(__PRETTY_FUNCTION__).append("\n");
            if (jobsQSem.tryAcquire())
            {
                util::SpinlockGuard guard(jobQLock);
                return jobQ.tryDequeue();
            }
            return NULL;
        }

        void enqueue(Actor *a)
        {
//          cout << std::string(__PRETTY_FUNCTION__).append("\n");
            jobQ.enqueue(a);
            jobsQSem.release();
        }

        // run the scheduler
        void operator()();

    } scheduler;

    struct Worker : public ReferenceCounted
    {
        volatile boost::uint32_t timestamp;
        volatile bool detached;
        volatile bool quit;
        volatile bool isRunning;
        volatile bool exited;
        boost::mutex mm; // only needed if we run a detached actor

        Actor *a;

        Worker() : ReferenceCounted(1), timestamp(0), detached(false),
                quit(false), isRunning(false), exited(false), a(NULL)
        {
        }

        Worker(Actor *detachedActor) : ReferenceCounted(1), timestamp(0),
                detached(true), quit(false), isRunning(false), exited(false),
                a(NULL)
        {
            assert(detachedActor->isDetached());
            a = detachedActor;
        }

        void waitUntilDetachedActorBecomesReady()
        {
            boost::unique_lock<boost::mutex> lock(mm);
            while (!a->m_lrmSet)
            {
                // check condition every 25ms again because there is a
                // possible way how lrmSet becomes true without notify
                boost::system_time timeout = boost::get_system_time();
                timeout += boost::posix_time::milliseconds(25);
                a->m_lrmSetCondition.timed_wait(lock, timeout);
            }
        }

        void runDetached()
        {
            Context ctx;
            assert(NULL != a && a->isDetached());
            // initialize if needed
            if (a->state() & UNINITIALIZED) a->initialize();
            // the state of the actor is detached
            // but the actor still yields READY, BLOCKED, etc.
            volatile ActorState nextState;
            a->yieldState = &nextState;
            while (!quit)
            {
                ACEDIA_MEMORY_BARRIER();
                a->resume(&ctx);
                ACEDIA_MEMORY_BARRIER();
                switch (nextState)
                {
                    case BLOCKED:
                        waitUntilDetachedActorBecomesReady();
                        break;

                    case READY: break; // run again

                    case EXITED:
                        // the exit operation may take some time,
                        // increase time stamp to prevent detaching
                        timestamp = timestamp + 1;
                        if (a->state() == DETACHED)
                        {
                            assert(a->setState(DETACHED, EXITED));
                            // do not decrease running actor count before
                            // the exit code has finished
                            ACEDIA_MEMORY_BARRIER();
                            decRunningActors();
                        }
                        // its either DETACHED or HIDDEN
                        else assert(a->setState(HIDDEN, EXITED));
                        // delete actor if needed
                        if (!a->deref()) delete a;
                        a = NULL;
                        quit = true; // cu
                        break;

                    default: throw 666; // must not happen
                }
            }
        }

        void operator()()
        {
            if (detached)
            {
                runDetached();
                exited = true;
                return;
            }
            ActorState nextState;
            Context ctx;
            while (!quit)
            {
                isRunning = false;
                ACEDIA_MEMORY_BARRIER();
                if (a)
                {
                    Actor *nextJob = scheduler.tryGetJob();
                    // did we become a new job?
                    if (nextJob)
                    {
                        // forsake the last job
                        scheduler.enqueue(a);
                        // and continue with the new one
                        a = nextJob;
                    }
                    // else: keep executing the current job
                }
                else
                {
                    // returns NULL if the scheduler has exited
                    a = scheduler.getJob();
                    if (!a)
                    {
                        quit = true;
                        ACEDIA_MEMORY_BARRIER();
                        exited = true;
                        return;
                    }
                }
                timestamp = timestamp + 1;
                // make sure timestamp is incremented before isRunning is set
                // to true
                ACEDIA_MEMORY_BARRIER();
                isRunning = true;
                if (a->state() & UNINITIALIZED) a->initialize();
                assert(!a->isDetached());
                a->yieldState = &nextState;
                assert(a->setState(READY, RUNNING));
                ACEDIA_MEMORY_BARRIER();
                a->resume(&ctx);
                ACEDIA_MEMORY_BARRIER();
                // did the actor consumed to much time?
                if (detached)
                {
                    if (nextState == EXITED)
                    {
                        assert(a->setState(RUNNING, EXITED));
                        if (!a->deref()) delete a;
                        a = NULL;
                    }
                    else
                    {
                        // make sure runDetached() sees a 'ready' actor
                        waitUntilDetachedActorBecomesReady();
                        runDetached();
                    }
                    decRunningActors();
                    exited = true;
                    return; // quit worker
                }
                else switch(nextState)
                {
                    case WAITING:
                    case BLOCKED:
                        assert(a->setState(RUNNING, nextState));
                        a = NULL; // don't continue this actor
                        break;

                    case READY:
                        assert(a->setState(RUNNING, READY));
                        break;

                    case EXITED:
                        // the exit operation may take some time,
                        // increase time stamp to prevent detaching
                        timestamp = timestamp + 1;
                        ACEDIA_MEMORY_BARRIER();
                        assert(a->setState(RUNNING, EXITED));
                        ACEDIA_MEMORY_BARRIER();
                        if (!a->deref()) delete a;
                        a = NULL;
                        decRunningActors();
                        break;

                    default: throw 666; // must not happen
                }
            }
            exited = true;
        }

    };

    struct WorkerRunner
    {
        Worker *w;
        WorkerRunner(Worker *worker) : w(worker) { if(w) w->ref(); }
        ~WorkerRunner()
        {
            if (w && !w->deref()) delete w;
        }
        WorkerRunner(const WorkerRunner &other) : w(other.w)
        {
            if (w) w->ref();
        }
        inline void operator()()
        {
            if (w) (*w)();
        }
    };

    SchedulerData::~SchedulerData()
    {
        if (worker && !worker->deref()) delete worker;
    }

    void Scheduler::operator()()
    {
        boost::uint32_t hc = boost::thread::hardware_concurrency();
        SchedulerData *data = new SchedulerData[hc];
        for (boost::uint32_t i = 0; i < hc; ++i)
        {
            Worker *w = new Worker;
            data[i].worker = w;
            boost::thread(WorkerRunner(w)).detach();
        }
        while (run)
        {
            for (boost::uint32_t i = 0; i < hc; ++i)
            {
                if (data[i].worker->isRunning)
                {
                    SchedulerData &d = data[i];
                    boost::uint32_t ts = d.worker->timestamp;
                    boost::system_time st = boost::get_system_time();
                    if (ts != d.lastReadedTStamp)
                    {
                        d.lastReadedTStamp = ts;
                        d.timeout = st;
                        // allow  a single actor to run 100ms
                        d.timeout += boost::posix_time::milliseconds(100);
                    }
                    else
                    {
                        // detach if needed
                        if (st >= d.timeout)
                        {
                            // detach the old one
                            d.worker->detached = true;
                            // and start a new one
                            d.lastReadedTStamp = UInt32Maximum;
                            d.worker = new Worker;
                            boost::thread(WorkerRunner(d.worker)).detach();
                        }
                    }
                }
            }
            // sleep for 1ms
            boost::system_time st = boost::get_system_time();
            st += boost::posix_time::milliseconds(1);
            boost::this_thread::sleep(st);
        }
        for (boost::uint32_t i = 0; i < hc; ++i)
        {
            data[i].worker->quit = true;
            ACEDIA_MEMORY_BARRIER();
            // wait until that worker finished work
            while (!data[i].worker->exited) boost::this_thread::yield();
        }
        delete[] data;
        ACEDIA_MEMORY_BARRIER();
        exited = true;
    }

    namespace details
    {
        void enqueueToScheduler(Actor *a) { scheduler.enqueue(a); }
    }

    void waitForAllActorsDone()
    {
        static boost::mutex mut;
        boost::unique_lock<boost::mutex> lock(mut);
        while (0 != runningActors)
        {
            // wakeup every 50 ms to check condition again
            boost::system_time timeout = boost::get_system_time();
            timeout += boost::posix_time::milliseconds(50);
            (void) noRunningActorsCv.timed_wait(lock, timeout);
        }
    }

    void shutdown()
    {
        scheduler.run = false;
        while (!scheduler.exited) boost::this_thread::yield();
#ifdef ACEDIA_WINDOWS
         WSACleanup();
#endif
    }

    namespace details
    {
        ActorRef doSpawn(Actor *a)
        {
            // there is an 'indirect' reference holded by the scheduler.
            // this reference is held until the actors state becomes EXITED.
            a->ref();
            if (a->isDetached())
            {
                // spawn a detached actor
                // start in a detached worker
                Worker *w = new Worker(a);
                boost::thread(WorkerRunner(w)).detach();
            }
            else
            {
                if (0 == schedulerStartetFlag
                    && 1 == atomic::addAndFetch(&schedulerStartetFlag, 1))
                {
#ifdef ACEDIA_WINDOWS
                    WSADATA wsa;
                    if (0 != WSAStartup(MAKEWORD(2,0), &wsa))
                    {
                        ACEDIA_THROW(WSAStartupFailedException, "WSAStartup(MAKEWORD(2,0), &wsa) failed");
                    }
#endif
                    boost::thread(boost::ref(scheduler)).detach();
                }
                ACEDIA_MEMORY_BARRIER();
                scheduler.enqueue(a);
            }
            if (!(a->state() & HIDDEN)) incRunningActors();
            return a->self();
        }
    }

} // namespace acedia

/******************************************************************************
 *                              acedia functions                              *
 ******************************************************************************/


static acedia::String m_application_name = "";
static acedia::String m_application_id = "(Version 0.0)";

static boost::uint32_t m_app_version_settings[4] = {0, 0, 0, 0};

static bool m_allowNewerMinorVersion = false;
static bool m_allowNewerMajorVersion = false;

static std::map<acedia::String, acedia::ActorRef> m_namesMap;
static boost::mutex m_namesMutex;

namespace acedia
{
    bool registerName(const String &name, const ActorRef &actor)
    {
        boost::mutex::scoped_lock lock(m_namesMutex);
        ActorRef &ar = m_namesMap[name];
        ar = actor;
        return true;
    }

    ActorRef whereisName(const String &name)
    {
        boost::mutex::scoped_lock lock(m_namesMutex);
        std::map<String, ActorRef>::const_iterator ci = m_namesMap.find(name);
        if (ci != m_namesMap.end())
        {
            return ci->second;
        }
        else if (name == "acedia_TimeEmitter")
        {
            ActorRef emitter = spawn<TimeEmitter>();
            m_namesMap.insert(std::map<String, ActorRef>::value_type(String("acedia_TimeEmitter"), emitter));
            return emitter;
        }
        else
        {
            return ActorRef();
        }
    }

    void setAppInfo(const char *appName,
                    boost::uint32_t majorVersion,
                    boost::uint32_t minorVersion,
                    boost::uint32_t minCompatibleMajorVersion,
                    boost::uint32_t minCompatibleMinorVersion,
                    bool allowNewerMinorVersion,
                    bool allowNewerMajorVersion)
    {
        m_application_name = appName;
        m_app_version_settings[0] = majorVersion;
        m_app_version_settings[1] = minorVersion;
        m_app_version_settings[2] = minCompatibleMajorVersion;
        m_app_version_settings[3] = minCompatibleMinorVersion;
        m_allowNewerMajorVersion = allowNewerMajorVersion;
        m_allowNewerMinorVersion = allowNewerMinorVersion;
        std::ostringstream os;
        os << appName << " (Version " << majorVersion << "."
           << minorVersion << ")";
        m_application_id = os.str();
    }
    namespace details
    {
        // appIdentifier = "{appName} (Version {major}.{minor})
        const String &appIdentifier()
        {
            return m_application_id;
        }
        const String &appName() { return m_application_name; }
        boost::uint32_t appMajorVersion()
        {
            return m_app_version_settings[0];
        }
        boost::uint32_t appMinorVersion()
        {
            return m_app_version_settings[1];
        }
        // return true if the appName corresponds to the set application name
        // and {major.minor} >= {minCompatibleMajor.minCompatibleMinor}
        bool isCompatible(const String& appn,
                          boost::uint32_t major,
                          boost::uint32_t minor)
        {
            return appn == appName()
                   && major >= m_app_version_settings[2]
                   && minor >= m_app_version_settings[3]
                   && ((m_allowNewerMajorVersion && major > appMajorVersion()) || major <= appMajorVersion())
                   && ((m_allowNewerMinorVersion && minor > appMinorVersion()) || minor <= appMinorVersion());
        }
    }
    void link(ActorRef actor1, ActorRef actor2)
    {
        actor1.actor->linkTo(actor2);
    }
}

---