acedia/acedia_details.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_DETAILS_H_
#define ACEDIA_DETAILS_H_

#include "acedia_config.hpp"

#include "unit.hpp"
#include "acedia_atomic.hpp"
#include "acedia_preprocessor.hpp"

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

namespace acedia
{

    // forward declaration
    class AnyArray;

    namespace details
    {

        // forward declaration
        class ContinuationInvoker;

        // Functor that takes an AnyArray and processes it.
        // Two or more processors may build a chain, so that functor1 calls functor2
        // etc.  To put a functor at the end of that chain use passThrough on the first
        // functor.
        // The chain is used to have the following format:
        // pattern -> additional checks (such as guards) -> invoker
        //
        // You can disable parts of the chain by set Operation.
        // With CHECK_ONLY you disable the invoker and with INVOKE_ONLY you disable
        // all preceding checks.
        class AnyArrayProcessor
        {

            public:

                AnyArrayProcessor();

                virtual ~AnyArrayProcessor();

                // return true if each processor evaluates to true
                // this is the case if the invoker "at the end" says true
                virtual bool operator()(const AnyArray &arr) = 0;

                // perform all checks and return:
                // - NULL if one of these checks fails
                // - an invoker if all checks evaluted to true
                virtual ContinuationInvoker *matchingInvoker(const AnyArray &arr) = 0;

                void passThrough(AnyArrayProcessor *nextProcessor);

                void setNext(AnyArrayProcessor *nextProcessor);

                inline bool hasNext() { return NULL != next; }

            protected:

                AnyArrayProcessor *next;

        };

        namespace container
        {
            template<typename T>
            T *inverseList(T *h)
            {
                if (!h) return NULL;
                // set head element
                T *n = h->next;
                h->next = NULL;
                T *inversedHead = h;
                // invert old head
                while (NULL != (h = n))
                {
                    n = h->next;
                    h->next = inversedHead;
                    inversedHead = h;
                }
                return inversedHead;
            }

            namespace intrusive
            {
                template<typename T>
                class Stack
                {
                    volatile T *head;

                public:

                    inline Stack() throw() : head(NULL) { }

                    void push(T *what)
                    {
                        for (;;)
                        {
                            T *h = const_cast<T*>(head);
                            what->next = h;
                            if (atomic::compareAndSwap(&head, h, what)) return;
                        }
                    }

                    T *pop()
                    {
                        for (;;)
                        {
                            T *h = const_cast<T*>(head);
                            if (h)
                            {
                                if (atomic::compareAndSwap(&head, h, h->next))
                                {
                                    return h;
                                }
                                // else: next for iteration
                            }
                            else return NULL;
                        }
                    }

                };

                template<typename T>
                class SingleReaderListBase
                {

                protected:

                    // the queue is implemented through a stack ...
                    volatile T *head;
                    // ... that is inversed by the reader
                    // to read the elements in FIFO order
                    T *inversedHead;

                    // take the first element from inversedHead and return it
                    T *takeFirstFromInversedHead()
                    {
                        if (inversedHead)
                        {
                            T *result = inversedHead;
                            inversedHead = inversedHead->next;
                            result->next = NULL;
                            return result;
                        }
                        return NULL;
                    }

                    // set head = NULL and return the old value of head
                    T *takeHead()
                    {
                        for (;;)
                        {
                            T *h = const_cast<T*>(head);
                            if (h)
                            {
                                if (atomic::compareAndSwap(&head, h,
                                            reinterpret_cast<T*>(NULL)))
                                {
                                    // successfully removed head from the queue
                                    // return old value
                                    return h;
                                }
                                // else: try again
                            }
                            else return NULL; // there is no head to remove
                        }
                    }

                    inline SingleReaderListBase() throw() :
                            head(NULL), inversedHead(NULL) { }

                    /*
                     * @brief Atomically enqueues @p what.
                     */
                    void append(T *what)
                    {
                        T *h;
                        do
                        {
                            h = const_cast<T*>(head);
                            what->next = h;
                        }
                        while (!atomic::compareAndSwap(&head, h, what));
                    }

                    /*
                     * @brief Dequeue an element.
                     *
                     * Note that this implementation is for single reader ONLY.
                     * This method does not lock at all and is absolutely not
                     * thread-safe in any way.
                     *
                     * @return NULL if there was no element;
                     *         otherwise the dequeued element
                     */
                    T *tryTakeFirst()
                    {
                        if (inversedHead) return takeFirstFromInversedHead();
                        else
                        {
                            inversedHead = inverseList(takeHead());
                            return takeFirstFromInversedHead();
                        }
                    }

                public:

                    inline bool isEmpty()
                    {
                        return NULL == head && NULL == inversedHead;
                    }

                    inline bool hasNext() { return !isEmpty(); }

                    void clear()
                    {
                        T *t;
                        while (NULL != (t = tryTakeFirst()))
                        {
                            delete t;
                        }
                    }

                };

                /*
                 * @brief SingleReaderQueue is a (intrusive and lock free)
                 *        Single Reader Many Writer Queue
                 *
                 * Complexity:<br>
                 * <ul>
                 * <li>enqueue: O(1)</li>
                 * <li>dequeue:<br>
                 *     - best case (mostly): O(1)<br>
                 *     - worst case: O(n)</li>
                 * </ul><br>
                 * This implementation uses the atomic CAS operations
                 * and does NOT use ANY lock.
                 *
                 * The queue is fast if there are at most only a few elements
                 * stored in the queue or if you have jerky a lot of enqueues
                 * followed by flushing.
                 *
                 * @p T must provide a <code>next</code> pointer which implies
                 *    that an instance of T can only be stored in ONE list.
                 */
                template<typename T>
                class SingleReaderQueue : public SingleReaderListBase<T>
                {

                public:

                    inline SingleReaderQueue() : SingleReaderListBase<T>() { }

                    /*
                     * @brief Atomically enqueues @p what.
                     */
                    inline void enqueue(T *what) { append(what); }

                    /*
                     * @brief Atomically enqueues a single linked list
                     *        described through @p lHead and @p lTail
                     */
                    void enqueueList(T *lHead, T *lTail)
                    {
                        T *h;
                        do
                        {
                            h = const_cast<T*>(this->head);
                            lTail->next = h;
                        }
                        while (!atomic::compareAndSwap(&(this->head), h,
                                                       lHead));
                    }

                    /*
                     * @brief Dequeue an element.
                     *
                     * Note that this implementation is for single reader ONLY.
                     * This method does not lock at all and is absolutely not
                     * thread-safe in any way.
                     *
                     * @return NULL if there was no element;
                     *         otherwise the dequeued element
                     */
                    inline T *tryDequeue() { return this->tryTakeFirst(); }

                };

                template<typename T>
                class SingleReaderList : public SingleReaderListBase<T>
                {

                    template<class C>
                    friend class SingleReaderListIterator;

                public:

                    T *front()
                    {
                        if (!(this->inversedHead))
                        {
                            this->inversedHead = inverseList(this->takeHead());
                        }
                        return this->inversedHead;
                    }

                    inline SingleReaderList() : SingleReaderListBase<T>() { }

                    inline T *tryTakeFirst()
                    {
                        return SingleReaderListBase<T>::tryTakeFirst();
                    }

                    inline void append(T *what)
                    {
                        SingleReaderListBase<T>::append(what);
                    }

                };

                template<class T>
                class SingleReaderListIterator
                {

                public:

                    typedef SingleReaderList<T> List;

                private:

                    List &list;
                    T *item;
                    T *prevItem;
                    bool initialized;

                public:

                    inline SingleReaderListIterator(List &parent) throw() :
                            list(parent), item(NULL), prevItem(NULL),
                            initialized(false) { }

                    inline T *next()
                    {
                        if (!initialized)
                        {
                            item = list.front();
                            initialized = true;
                        }
                        else
                        {
                            if (item)
                            {
                                prevItem = item;
                                if (!item->next)
                                {
                                    item->next = inverseList(list.takeHead());
                                }
                                item = item->next;
                            }
                        }
                        return item;
                    }

                    // erase current item and return the erased item
                    // note: this also sets the iterator to next()
                    T *erase()
                    {
                        if (!item) return NULL;
                        T *i = item;
                        if (!prevItem)
                        {
                            // we delete the first item in the list
                            list.inversedHead = item->next;
                        }
                        else
                        {
                            prevItem->next = item->next;
                            item = prevItem->next;
                        }
                        return i;
                    }

                    inline T &operator*() { return *item; }

                    inline bool hasNext()
                    {
                        if (initialized)
                        {
                            if (!item) return false;
                            if (!item->next)
                            {
                                item->next = inverseList(list.takeHead());
                                return NULL != item->next;
                            }
                            return true;
                        } else {
                            if (!list.inversedHead)
                            {
                                list.inversedHead =
                                        inverseList(list.takeHead());
                            }
                            return NULL != list.inversedHead;
                        }
                    }

                    inline bool isValid() { return NULL != item; }

                };

                /*
                 * @brief Many Reader Many Writer Queue
                 *
                 * Queue itself is not lock free, but the enqueue operation is.
                 * A Lock is used to synchronize dequeue operations.
                 *
                 * Queue has the same complexity as SingleReaderQueue (in fact
                 * it uses a SingleReaderQueue in its implementation).
                 */
                template<typename T>
                class Queue
                {

                    SingleReaderQueue<T> impl;
                    boost::condition_variable cv;
                    boost::mutex mm;

                public:

                    // nonblocking!
                    void enqueue(T *what)
                    {
                        impl.enqueue(what);
                        ACEDIA_MEMORY_BARRIER();
                        cv.notify_all();
                    }

                    // blocking!
                    T *dequeue()
                    {
                        boost::unique_lock<boost::mutex> lock(mm);
                        // to READ an int32 is atomic, so there's no need
                        // for atomic::read() or something
                        T *t = impl.tryDequeue();
                        while (!t)
                        {
                            cv.wait(lock);
                            t = impl.tryDequeue();
                        }
                        return t;
                    }

                    // blocking!
                    T *tryDequeue()
                    {
                        boost::unique_lock<boost::mutex> lock(mm);
                        return impl.tryDequeue();
                    }

                    // blocking!
                    T *tryDequeue(boost::int64_t msTimeout)
                    {
                        boost::system_time timeout = boost::get_system_time();
                        timeout += boost::posix_time::milliseconds(msTimeout);
                        boost::unique_lock<boost::mutex> lock(mm);
                        T *t = impl.tryDequeue();
                        while (!t)
                        {
                            if (!cv.timed_wait(lock, timeout)) return NULL;
                            t = impl.tryDequeue();
                        }
                        return t;
                    }

                };

            } // namespace intrusive

            /*
             * T must offer a copy constructor.
             */
            template<typename T>
            struct Node
            {
                T value;
                Node *next;
                inline Node(const T &t) : value(t), next(NULL) { }
            private:
                // no copy construction and no assignment
                Node(const Node &other);
                Node &operator=(const Node &other);
            };

            template<typename T>
            class SingleReaderQueue
            {

            private:

                intrusive::SingleReaderQueue< Node<T> > impl;

            public:

                inline void enqueue(const T &what)
                {
                    impl.enqueue(new Node<T>(what));
                }

                inline void enqueueList(Node<T> *lHead, Node<T> *lTail)
                {
                    impl.enqueueList(lHead, lTail);
                }

                bool tryDequeue(T &storage)
                {
                    Node<T> *n = impl.tryDequeue();
                    if (n)
                    {
                        storage = n->value;
                        delete n;
                        return true;
                    }
                    return false;
                }

                inline Node<T> *tryDequeueNode() { return impl.tryDequeue(); }

                inline bool isEmpty() { return impl.isEmpty(); }

                inline bool hasNext() { return !isEmpty(); }

            };

            template<typename T>
            class Stack
            {

            intrusive::Stack< Node<T> > impl;

            public:

                void push(const T &what)
                {
                    impl.push(new Node<T>(what));
                }

                bool pop(T &storage)
                {
                    Node<T> *n = impl.pop();
                    if (n)
                    {
                        storage = n->value;
                        delete n;
                        return true;
                    }
                    return false;
                }

            };

            template<typename T>
            class SingleReaderList
            {

                template<typename C>
                friend class SingleReaderListIterator;

            public:

                typedef Node<T> ListNode;

            private:

                intrusive::SingleReaderList<ListNode> impl;

            public:

                inline void append(const T &what)
                {
                    impl.append(new ListNode(what));
                }

                bool tryTakeFirst(T &storage)
                {
                    ListNode *n = impl.tryTakeFirst();
                    if (n)
                    {
                        storage = n->value;
                        delete n;
                        return true;
                    }
                    return false;
                }

                inline bool isEmpty() { return impl.isEmpty(); }

                inline bool hasNext() { return impl.hasNext(); }

                inline void clear() { impl.clear(); }

            };

            template<typename T>
            class SingleReaderListIterator
            {

            public:

                typedef Node<T> ListNode;
                typedef SingleReaderList<T> List;

            private:

                intrusive::SingleReaderListIterator<ListNode> i;

            public:

                inline SingleReaderListIterator(List &parent) :
                        i(parent.impl) { }

                inline SingleReaderListIterator(List &parent, ListNode *n) :
                        i(parent.impl, n) { }

                inline void operator++() { ++i; }

                inline T &operator*() { return (*i).value; }

                inline bool isValid() { return i.isValid(); }

                inline T &next() { return (i.next())->value; }

                inline bool hasNext() { return i.hasNext(); }

                inline void erase()
                {
                    ListNode *n = i.erase();
                    if (n) delete n;
                }

            };

            /*
             * T must offer a copy constructor.
             */
            template<typename T>
            class Queue
            {

                intrusive::Queue< Node<T> > impl;

                bool setStorageIfN(T &storage, Node<T> *n)
                {
                    if (n)
                    {
                        storage = n->value;
                        delete n;
                        return true;
                    }
                    return false;
                }

            public:

                // nonblocking!
                inline void enqueue(const T &what)
                {
                    impl.enqueue(new Node<T>(what));
                }

                // blocking!
                inline void dequeue(T &storage)
                {
                    (void) setStorageIfN(storage, impl.dequeue());
                }

                // blocking!
                inline bool tryDequeue(T &storage)
                {
                    return setStorageIfN(storage, impl.tryDequeue());
                }

                // blocking!
                inline bool tryDequeue(T &storage, boost::int64_t msTimeout)
                {
                    return setStorageIfN(storage, impl.tryDequeue(msTimeout));
                }

            };

        } // namespace container

    } // namespace details

} // namespace acedia

#endif

---