include/amino/queue_shavit.h

Go to the documentation of this file.

/*
 * (c) Copyright 2008, IBM Corporation.
 *
 *      Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#ifndef QUEUE_SHAVIT_H_
#define QUEUE_SHAVIT_H_
#include "smr.h"
#include <iostream>

namespace amino {
using namespace internal;

template<typename T> class ShavitQueue {
private:
        class QueueItem {
        public:
                T data;
                QueueItem* next;
                QueueItem* prev;

                QueueItem() :
                        data() {
                        next = NULL;
                        prev = NULL;
                }

                QueueItem(const T& val) :
                        data(val) {
                        next = NULL;
                        prev = NULL;
                }

                QueueItem(QueueItem* n) :
                        data() {
                        next = n;
                        prev = NULL;
                }

                inline QueueItem* getNext() {
                        return prev;
                }
        };

        /*
         * Cannot be static since dummy->next is different in different queue objects.
         * But one object need one dummy node only.
         */
        QueueItem *dummy;

        atomic<QueueItem *> head;
        atomic<QueueItem *> tail;
        /*
         * SMR for memory management. All objects of the class queue share one SMR.
         */
        SMR<QueueItem, 2>* mm;
    typedef typename SMR<QueueItem, 2>::HP_Rec HazardP;

        static const bool hasBackoff = false;

        /* prevent the use of copy constructor and copy assignment*/
        ShavitQueue(const ShavitQueue& q) {
        }

        ShavitQueue& operator=(const ShavitQueue& q) {
        }

        void fixList(QueueItem *tail, QueueItem *head) {
                QueueItem *curNode, *curNodeNext, *nextNodePrev;
                curNode = tail;
                while ((this->head.load(memory_order_relaxed) == head) && (curNode
                                != head)) {
                        curNodeNext = curNode->next;
                        if (NULL == curNodeNext) { /*ABA, return*/
                                assert(false);
                        }
                        nextNodePrev = curNodeNext->prev;
                        if (nextNodePrev != curNode) {
                                curNodeNext->prev = curNode;
                        }
                        curNode = curNodeNext;
                }
        }
public:
        ShavitQueue() {
                mm = getSMR<QueueItem, 2> ();
                dummy = new QueueItem();
                head.store(dummy, memory_order_relaxed);
                tail.store(dummy, memory_order_relaxed);
        }

        ~ShavitQueue() {
                QueueItem *first = head.load(memory_order_relaxed);
                if (first == dummy) {
                        first = first->prev;
                }
                while (tail.load(memory_order_relaxed) != dummy) {
                        if (first == tail.load(memory_order_relaxed)) {
                                delete first;
                                break;
                        }
                        QueueItem *prev = first -> prev;
                        delete first;
                        first = prev;
                }

                delete dummy;
        }

        void enqueue(const T& x) {
        HazardP * hp = mm->getHPRec();
#ifdef FREELIST
                QueueItem *node = mm->newNode(hp); /*Allocate a new node*/
        node->data =x;
#else
                QueueItem *node = new QueueItem(x);/*Allocate a new node*/
#endif

                int wait_for_backoff = 10000;
                while (true) {
            QueueItem * t = tail.load(memory_order_relaxed); /*Get the tail*/

                        /*hazard*/
            //FIXME hp as first parameter?
                        mm->employ(hp,0, t);
                        /*If the tail be modified, start over from the beginning of the loop*/
                        if (tail.load(memory_order_relaxed) != t) {
                                continue;
                        }

                        node->next = t;/*Set the node's next pointer*/
                        if (tail.compare_swap(t, node)) {/*try to swing tail to the new node. If success, set the previous pointer*/
                                t->prev = node;
                                break;
                        }

                        if (hasBackoff) {
                                usleep(wait_for_backoff);
                                wait_for_backoff <<= 1;
                        }
                }
        }

        bool dequeue(T& val) {
                QueueItem *tl, *hd, *fstNodePrev;

        HazardP * hp = mm->getHPRec();

                int wait_for_backoff = 10000;
                while (true) {/*Try till success or empty*/
                        hd = head.load(memory_order_relaxed);

                        /*hazard*/
                        mm->employ(hp, 0, hd);
                        /*If the tail be modified, start over from the beginning of the loop*/
                        if (head.load(memory_order_relaxed) != hd) {
                                continue;
                        }

                        tl = tail.load(memory_order_relaxed);

                        fstNodePrev = hd->prev;

                        /*hazard*/
                        mm->employ(hp, 1, fstNodePrev);
                        if (hd->prev != fstNodePrev) {
                                continue;
                        }

                        val = hd->data;

                        if (head.load(memory_order_relaxed) == hd) {/*Check consistency*/
                                /*
                                 * Head value is dummy?
                                 */
                                if (hd != dummy) {
                                        if (tail.load(memory_order_relaxed) != head.load(
                                                        memory_order_relaxed)) { /*More than 1 node?*/
                                                if (NULL == fstNodePrev) {
                                                        fixList(tl, hd);
                                                        continue;
                                                }
                                        } else {/*Last node in queue*/
                                                dummy->next = tl;
                                                dummy->prev = NULL;
                                                if (tail.compare_swap(tl, dummy)) {/*CAS tail*/
                                                        hd->prev = dummy;
                                                }
                                                continue;
                                        }
                                        /*CAS head*/
                                        if (head.compare_swap(hd, fstNodePrev)) {
                                                //FIXME free(head) free the dequeued node
                                                /* need SMR here */
                                                mm->delNode(hp, hd);
                                                return true;
                                        }
                                } else {/*Head points to dummy*/
                                        if (tl == hd) {/*Tail points to dummy?*/
                                                //empty queue
                                                return false;
                                        } else {/*Need to skip dummy*/
                                                if (NULL == fstNodePrev) {/*prev pointer is inconsistency*/
                                                        fixList(tl, hd);
                                                        continue;
                                                }
                                                /*Skip dummy*/
                                                head.compare_swap(hd, fstNodePrev);
                                        }
                                }
                        }
                        if (hasBackoff) {
                                usleep(wait_for_backoff);
                                wait_for_backoff <<= 1;
                        }
                }
        }

        bool empty() {
                return head.load(memory_order_relaxed) == dummy && head.load(memory_order_relaxed) == tail.load(memory_order_relaxed);
        }

        int size() {
                int ret = 0;
                QueueItem *first = head.load(memory_order_relaxed);
                if (first == dummy) {
                        first = first->prev;
                }
                while (tail.load(memory_order_relaxed) != dummy) {
                        ++ret;
                        if (first == tail.load(memory_order_relaxed))
                                break;
                        first = first->prev;
                }
                return ret;
        }

        bool peekFront(T& ret) {
                while (true) {
                        QueueItem *front = head.load(memory_order_relaxed);
                        ret = front->data;
                        if (dummy != front) {
                                return true;
                        }

                        QueueItem *end = tail.load(memory_order_relaxed);
                        if (front == end) {
                                return false;
                        } else {
                                QueueItem *fstNodePrev = front->prev;
                                if (NULL == fstNodePrev) {
                                        fixList(end, front);
                                        continue;
                                }
                                head.compare_swap(front, fstNodePrev);
                        }
                        return true;
                }
        }
};
}
#endif /*QUEUE_SHAVIT__H_*/

---