OgreHapticsConcurrentQueue.h

Go to the documentation of this file.

/*
This source file is part of OgreHaptics
 (a library for wrapping haptics devices for use with the open-source
 graphics engine OGRE, http://www.ogre3d.org)

Copyright (c) 2006 - 2008 Jorrit de Vries
Also see acknowledgements in Readme.html

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 2.1
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 this library; if not, write to the
Free Software Foundation, Inc.,
59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/
#ifndef _OgreHapticsConcurrentQueue_H__
#define _OgreHapticsConcurrentQueue_H__

#include "OgreHapticsPrerequisites.h"
#include "OgreHapticsAtomic.h"
#include "OgreHapticsNonCopyable.h"

namespace OgreHaptics {

    template<typename T> class ConcurrentQueue : public NonCopyable
    {
    public:
        // Predeclaration
        class HazardPtr;
    private:
        // Structure which is used to contain data within the queue.
        struct Node
        {
            T data;
            Node* volatile next;

            Node()
                : data(), next(0) {}
            Node(T val)
                : data(val), next(0) {}
        };

        // Head of single linked list of hazard pointers.
        HazardPtr* mHPHead;
        // Number of hazards present at most in list of hazard pointers.
        int mHPCount;
        // Head of queue.
        Node* volatile mHead;
        // Tail of queue.
        Node* volatile mTail;

        typedef std::vector<Node*> NodeList;
    public:
        class HazardPtr : public NonCopyable
        {
            // Allow ConcurrentQueue full access.
            friend class ConcurrentQueue;
        public:
            HazardPtr()
                : active(1), 
                next(0)
            {
                hazards[0] = 0;
                hazards[1] = 0;
            }
            virtual ~HazardPtr() {}
        protected:
            int active;
            HazardPtr* next;
            Node* hazards[2];
            NodeList retiredList;
        };

        ConcurrentQueue()
            : mHPHead(0),
            mHPCount(0)
        {
            // Create dummy node ensuring both head and tail point at a node
            // in the queue to avoid problems when the queue is empty or
            // contains only a single item.
            mHead = mTail = new Node();
        }

        ~ConcurrentQueue()
        {
            // Cleanup list of hazard pointers
            while (mHPHead)
            {
                // Advance in list
                HazardPtr* tmp = mHPHead;
                mHPHead = mHPHead->next;

                // Clean up left retired nodes
                NodeList::iterator it;
                for (it = tmp->retiredList.begin(); it != tmp->retiredList.end(); ++it)
                {
                    delete *it;
                }
                tmp->retiredList.clear();

                delete tmp;
            }

            // Cleanup head (head is actually never removed)
            delete mHead;
        }

        HazardPtr* allocateHazardPtr(void)
        {
            // First try to use a retired hazard pointer
            HazardPtr* ptr = mHPHead;
            for (; ptr; ptr = ptr->next)
            {
                if (ptr->active != 0 || !atomicCompareAndSwap(&ptr->active, 0, 1))
                {
                    continue;
                }
                return ptr;
            }

            // No hazard pointer available for reuse
            // Increment list count by two (maximum two hazards per thread)
            int count;
            do
            {
                count = mHPCount;
            }
            while (!atomicCompareAndSwap(&mHPCount, count, count + 2));

            // Allocate a new hazard pointer
            ptr = new HazardPtr();
            // Push new hazard pointer to front
            HazardPtr* oldHead;
            do
            {
                oldHead = mHPHead;
                ptr->next = oldHead;
            }
            while (!atomicCompareAndSwapPtr(&mHPHead, oldHead, ptr));

            return ptr;
        }

        void retireHazardPtr(HazardPtr* ptr)
        {
            // Clear hazards
            ptr->hazards[0] = 0;
            ptr->hazards[1] = 0;
            // Deactivate hazard pointer after ensuring previous operations
            // have been executed (compare-and-swap issues full memory barrier)
            atomicCompareAndSwap(&ptr->active, 1, 0);
            return;
        }

        void push(HazardPtr* ptr, const T& value)
        {
            assert((ptr && ptr->active != 0) && "HazardPtr must be active.");

            // Create new node
            Node* node = new Node(value);
            Node* tail;

            while (true)
            {
                tail = mTail;
                // Hazard, tail may have been removed by other thread after
                // previous line has been executed
                ptr->hazards[0] = tail;
                // Some other thread might have removed tail and didn't find
                // hazardous references to it
                if (tail != mTail)
                {
                    continue;
                }
                Node* next = tail->next;
                // We need to guarantee next is from actual tail in order to
                // prevent ABA problems
                if (tail != mTail)
                {
                    continue;
                }
                if (next)
                {
                    // Keep updating tail until it points to last node
                    atomicCompareAndSwapPtr(&mTail, tail, next);
                    continue;
                }
                // If tail points to last node update tail to point to new node
                if (atomicCompareAndSwapPtr(&tail->next, reinterpret_cast<Node*>(0), node))
                {
                    break;
                }
            }

            // If tail points to what we thought was last node update tail to
            // point at new node
            atomicCompareAndSwapPtr(&mTail, tail, node);
        }

        bool pop(HazardPtr* ptr, T& destination)
        {
            assert((ptr && ptr->active != 0) && "HazardPtr must be active.");

            Node* head;
            
            while (true)
            {
                head = mHead;
                ptr->hazards[0] = head;
                if (head != mHead)
                {
                    continue;
                }

                Node* tail = mTail;

                Node* next = head->next;
                ptr->hazards[1] = next;
                if (head != mHead)
                {
                    continue;
                }

                // Queue is empty
                if (!next)
                {
                    return false;
                }

                if (head == tail)
                {
                    // Tail is behind, move tail off of head
                    atomicCompareAndSwapPtr(&mTail, tail, next);
                    continue;
                }
                else
                {
                    destination = next->data;
                    // Move head pointer, effectively removing the node
                    if (atomicCompareAndSwapPtr(&mHead, head, next))
                    {
                        break;
                    }
                }
            }

            // Mark node for reclamation
            retireNode(ptr, head);
            return true;
        }
    private:
        void retireNode(HazardPtr* ptr, Node* node)
        {
            // Add node to retiredList of hazard pointer
            ptr->retiredList.push_back(node);
            HazardPtr* head = mHPHead;
            
            // Check whether we need to scan for nodes eligible for reclamation
            //TODO comment omega 
            static const float omega = 1.25f;
            if (ptr->retiredList.size() >= (omega * mHPCount))
            {
                scan(ptr, head);
                helpScan(ptr);
            }
        }

        void scan(HazardPtr* ptr, HazardPtr* head)
        {
            // Scan list of hazard pointers to collect all non-null hazards
            NodeList hazards;
            while (head)
            {
                for (int i = 0; i < 2; ++i)
                {
                    Node* hazard = head->hazards[i];
                    if (hazard)
                    {
                        hazards.push_back(hazard);
                    }
                }
                head = head->next;
            }

            // Sort the list of hazards
            std::sort(hazards.begin(), hazards.end(), std::less<Node*>());

            // Check each hazard against retired nodes
            NodeList retiredList(ptr->retiredList);
            ptr->retiredList.clear();
            NodeList::iterator it;
            for (it = retiredList.begin(); it != retiredList.end(); ++it)
            {
                if (!std::binary_search(hazards.begin(), hazards.end(), *it))
                {
                    // Node is eligible for reuse
                    delete *it;
                }
                else
                {
                    ptr->retiredList.push_back(*it);
                }
            }
        }

        void helpScan(HazardPtr* ptr)
        {
            HazardPtr* head = mHPHead;
            for (; head; head = head->next)
            {
                if (head->active != 0 || !atomicCompareAndSwap(&head->active, 0, 1))
                {
                    // Only scan inactive hazard pointers. Inactive hazard
                    // pointers are locked through compare-and-swap
                    continue;
                }
                while (!head->retiredList.empty())
                {
                    // Add node to list owned by given hazard pointer
                    ptr->retiredList.push_back(head->retiredList.back());
                    head->retiredList.pop_back();
                    HazardPtr* h = mHPHead;
                    //TODO comment omega
                    static const float omega = 1.25f;
                    if (ptr->retiredList.size() > (omega * mHPCount))
                    {
                        scan(ptr, h);
                    }
                }
                // Deactivate hazard pointer
                head->active = 0;
            }
        }
    };

}

#endif

---

Last modified Tue Jan 6 22:31:25 2009