root/trunk/tango/scrapple/thread/Actor.d

/**
 * An implementation for an actor in the actor model of concurrency
 * 
 * Authors:
 *  Gregor Richards
 * 
 * License:
 *  Copyright (c) 2007  Gregor Richards
 *  
 *  Permission is hereby granted, free of charge, to any person obtaining a
 *  copy of this software and associated documentation files (the "Software"),
 *  to deal in the Software without restriction, including without limitation
 *  the rights to use, copy, modify, merge, publish, distribute, sublicense,
 *  and/or sell copies of the Software, and to permit persons to whom the
 *  Software is furnished to do so, subject to the following conditions:
 *  
 *  The above copyright notice and this permission notice shall be included in
 *  all copies or substantial portions of the Software.
 *  
 *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 *  FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 *  DEALINGS IN THE SOFTWARE.
 */

module tango.scrapple.thread.Actor;

public import tango.scrapple.thread.Message;

public import tango.core.Thread;
import tango.core.sync.Mutex;
import tango.core.sync.Semaphore;

/** An actor is a thread with builtin message-passing support for concurrency */
class Actor : Thread {
    public:
        this()
        {
            actorLock = new Mutex;
            queueLock = new Semaphore;
            super(&run);
        }

        /** new with a master */
        this(Actor smaster)
        {
            actorLock = new Mutex;
            queueLock = new Semaphore;
            master = smaster;
            super(&run);
        }

        /** Enqueue a message */
        void enqueue(Message m)
        {
            // before locking, try to give the recipient a fair chance to receive
            Thread.yield();

            // lock the queue
            actorLock.lock();

            // add the new message
            queue ~= m;

            // then notify of the queue update
            queueLock.notify();

            // and unlock the queue
            actorLock.unlock();
        }

        /** Enqueue a message and wait (must be done from an actor).
            Returns the message. */
        Message enqueueSync(Message m)
        {
            // create a semaphore to wait for the synchronization on
            scope Semaphore s = new Semaphore;

            // now enqueue it
            enqueue(new SyncMessage(s, m));

            // and wait
            s.wait();

            return m;
        }

        /** Enqueue a group of messages and wait for all of them (really only
            useful with a Dispatcher or some other delegating actor) */
        void enqueueSync(Message[] msgs)
        {
            // make all our semaphores
            Semaphore[] sems;
            sems.length = msgs.length;
            foreach (i, m; msgs) {
                sems[i] = new Semaphore;
            }

            // enqueue all the messages
            foreach (i, m; msgs) {
                enqueue(new SyncMessage(sems[i], m));
            }

            // then wait on all the semaphores
            foreach (s; sems) {
                s.wait();
            }
        }

        /** Enqueue a delegate to be called */
        void enqueueDelegate(void delegate() d)
        {
            enqueueSync(new DelegateMessage(d));
        }

        /** You can also enqueue a delegate by just telling an actor to "do"
            something */
        alias enqueueDelegate act;

        /** Set our master */
        void setMaster(Actor smaster)
        {
            master = smaster;
        }

        /** Handle a message. The default handler is to call the Message's own
            handling function. Returns true if this actor should die. */
        bool handle(Message m)
        {
            return m.handle(this);
        }

    private:
        /** The basic actor loop */
        void run()
        {
            while (true) {
                actorLock.lock();

                // check the queue
                while (queue.length != 0) {
                    // get the message off the head
                    Message qhead = queue[0];

                    /* move everything else back (slicing would cause
                       unnecessary overhead when enqueueing) */
                    for (int i = 0; i < queue.length - 1; i++) {
                        queue[i] = queue[i+1];
                    }
                    queue.length = queue.length - 1;
                    
                    // then run it (allowing other messages to come in)
                    actorLock.unlock();
                    if (handle(qhead)) {
                        // that's it for us!
                        return;
                    }
                    actorLock.lock();

                    // empty queue! Let's complain about it
                    if (queue.length == 0 && master !is null) {
                        actorLock.unlock();
                        master.enqueue(new EmptyQueueMessage(this));
                        actorLock.lock();
                    }
                }

                // now we're just waiting on new data
                actorLock.unlock();

                queueLock.wait();
            }
        }

        /** The actor lock. Cannot generally be used through synchronize() */
        Mutex actorLock;

        /** The queue lock informs the thread when new items have entered the
            queue */
        Semaphore queueLock;

        /** The queue of commands. Should only be accessed through the proper
            locking mechanisms */
        Message[] queue;

        /** The master, to which messages such as EmptyQueueMessage are sent */
        Actor master;
}

/** A Dispatcher is an actor which delegates messages to other actors. Useful
    when you have a number of actors for different cores/CPUs, but they are
    otherwise indistinct */
class Dispatcher : Actor {
    public:
        this(Actor[] sslaves)
        {
            super();

            slaves = sslaves.dup;
            
            // mark this as the master of each slave
            foreach (s; slaves) {
                s.setMaster(this);
                slaveQueueEmpty[s] = true;
            }
        }

        /** Handle messages */
        bool handle(Message m)
        {
            if (cast(EmptyQueueMessage) m !is null) {
                // just telling us their queue is empty, so mark it
                if (m.sender !is null)
                    slaveQueueEmpty[m.sender] = true;

            } else if (cast(DieMessage) m !is null) {
                // when we die, we take everyone down with us!
                foreach (s; slaves)
                    s.enqueueSync(new DieMessage);
                return true;

            } else if (auto sm = cast(SyncMessage) m) {
                // this could be a DieMessage
                auto subm = sm.m;
                if (cast(DieMessage) subm !is null) {
                    foreach (s; slaves)
                        s.enqueueSync(new DieMessage);
                    return m.handle(this);
                }
                delegateQueue ~= m;

            } else {
                delegateQueue ~= m;

            }

            flushDelegateQueue();
            return false;
        }

    private:
        /** Flush the delegate queue as much as possible */
        void flushDelegateQueue()
        {
            foreach (a; slaves) {
                if (slaveQueueEmpty[a]) {
                    // give them something to do
                    if (delegateQueue.length > 0) {
                        a.enqueue(delegateQueue[0]);
                        for (int i = 0; i < delegateQueue.length - 1; i++) {
                            delegateQueue[i] = delegateQueue[i+1];
                        }
                        delegateQueue.length = delegateQueue.length - 1;
                        slaveQueueEmpty[a] = false;
                    }
                }
            }
        }

        /** The slaves of this delegater */
        Actor[] slaves;

        /** And whether their queues are empty */
        bool[Actor] slaveQueueEmpty;

        /** A separate queue for messages which we have received, but not yet
            delegated */
        Message[] delegateQueue;
}