queue.cpp

// Copyright (C) Calum Grant 2008

#include "cppscript"
#include "dynamic/extensions.hpp"
#include "thread.hpp"


namespace
{
        namespace Queue
        {
                var pop(var queue)
                {
                        {
                                internal::leave_apartment l;
                                queue["Queue::num_waiting"].as<api::stack>().wait_pop();
                        }
                        return queue["Queue::pending"].pop_front();
                }

                void post(var queue, var message)
                {
                        queue["Queue::pending"].push_back(message);
                        queue["Queue::num_waiting"].as<api::stack>().push();
                }

                void wait(var queue)
                {
                        var num_waiting = queue["Queue::num_waiting"];
                        internal::leave_apartment l;
                        num_waiting.as<api::stack>().wait_empty();
                }
        }


        namespace MessageQueue
        {
                void post(var queue, var message)
                {
                        Queue::post(queue, message);
                        queue["MessageQueue::num_done"].as<api::stack>().push();
                }

                void process(var queue)
                {
                        var item = queue["pop"]();

                        try
                        {
                                queue["consumer"](item);
                        }
                        catch(...)
                        {
                                // Silently ignore exceptions (not good design)
                        }

                        internal::leave_apartment l;
                        queue["MessageQueue::num_done"].as<api::stack>().wait_pop();
                }

                void process_thread(var queue)
                {
                        while( !queue["Queue::num_waiting"].as<api::stack>().check_shutdown() )
                        {
                                process(queue);
                        }
                }

                void close(var queue)
                {
                        queue["Queue::num_waiting"].as<api::stack>().shutdown();
                        queue["MessageQueue::num_done"].as<api::stack>().shutdown();

                        foreach(thread, queue["threads"])
                                thread["join"]();
                }

                void wait(var queue)
                {
                        Queue::wait(queue);
                        var num_done = queue["MessageQueue::num_done"];
                        internal::leave_apartment l;
                        num_done.as<api::stack>().wait_empty();
                }
        }
}


var dynamic::queue()
{
        return object("Queue").extend
                ("Queue::pending", list())
                ("post", Queue::post)
                ("Queue::num_waiting", create<api::stack>())
                ("pop", Queue::pop)
                ("wait", Queue::wait);
}


var dynamic::message_queue(const var & consumer, const var & num_threads)
{
        var q = queue().extend
                ("process_thread", MessageQueue::process_thread)
                ("consumer", consumer)
                ("process", MessageQueue::process)
                ("threads", array())
                ("close", MessageQueue::close)
                ("post", MessageQueue::post)
                ("MessageQueue::num_done", create<api::stack>())
                ("wait", MessageQueue::wait);

        for(var i = 0; i<num_threads; ++i)
        {
                q["threads"].push_back(thread(q["process_thread"]));
        }

        return q;
}

---