include/amino/deque.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.
 *
 *  Change History:
 *
 *  yy-mm-dd  Developer  Defect     Description
 *  --------  ---------  ------     -----------
 *  08-08-03  ganzhi     N/A        Initial implementation
 */

#ifndef AMINO_DEQUE_H
#define AMINO_DEQUE_H

#include <stdexcept>

#include <amino/util.h>
#include <amino/smr.h>

#ifdef DEBUG
#include <amino/debug.h>
#endif

namespace amino {
        /* needed by SMR */
        using namespace internal;

        /*
         * A deque is stable only if the status tag holds the value stable.
         * A deque can be stable only if it is coherent. That is, for all
         * nodes x in the deque's list x.Right.Left=x, unless x is the
         * rightmost node; and x.Left.Right=x, unless x is the leftmost node.
         *  Empty and single-item deques are always stable.A deque can be
         *  unstable only if it contains two or more items. It is acceptable
         *  for one end of the deque to be incoherent, as long as the status
         *  tag indicates an unstable deque on the same end. The list,
         *  excluding the end nodes, is always coherent. The deque is
         *  right-incoherent if r.Left.Right=r, where r is the rightmost node.
         *  The deque is left-incoherent if l.Right.Left=l, where l is the
         *  leftmost node. And the status tag is rpush and lpush respectively.
         *  At most one end of the deque can be incoherent at a time.
         *  Unstable states result from push operations, but never from pop
         *  operations, hence the aming of the unstable status tag values.
         *  It is also acceptable for the status tag to indicate an unstable
         *  deque even when the deque is in fact coherent.
         */
#define STABLE 0
#define RPUSH 1
#define LPUSH 2

        template<typename T>
        class LockFreeDeque {
        private:
                LockFreeDeque(const LockFreeDeque& dq) {
                }

                LockFreeDeque& operator=(const LockFreeDeque& dq) {
                }

                class Node {
                public:
                        T data;

                        atomic<Node*> left;

                        atomic<Node*> right;

                        Node() :
                        data() {
                                left.store(NULL, memory_order_relaxed);
                                right.store(NULL, memory_order_relaxed);
                        }

                        Node(const T& val) :
                        data(val) {
                                left.store(NULL, memory_order_relaxed);
                                right.store(NULL, memory_order_relaxed);
                        }
                };

                typedef union AnchorType {
                        struct {
                                volatile unsigned long low;
                                volatile unsigned long high;
                        }all;

                        struct {
                                /* pointer to left nodes, 32 bits */
                                volatile unsigned long left;

                                /* pointer to right nodes. 30 bits.
                                 * The LSB(two bits) is used by <code>status</code> field.
                                 */
                                volatile unsigned long right :(sizeof(unsigned long) * 8 - 2);

                                /* status of the Anchor. 2 bits.
                                 * it can be one of the "STABLE", "LPUSH", "RPUSH"
                                 */
                                volatile unsigned long status :2;//status type
                        }fields;

                public:
                        void setRight(Node * pNode) {
                                fields.right = ((unsigned long) pNode) >> 2;
                                assert(getRight() == pNode);
                        }

                        Node * getRight() {
                                return (Node *) (fields.right << 2);
                        }

                        Node * getLeft() {
                                return (Node *) fields.left;
                        }
#if defined(BIT32)
                }__attribute__((aligned(8))) Anchor_t;
#elif defined(BIT64)
        }__attribute__((aligned(16))) Anchor_t;
#endif

        SMR<Node, 3>* mm; /* SMR for memory management */
        typedef typename SMR<Node, 3>::HP_Rec HazardP;
        volatile Anchor_t anchor;

        void copyAnchor(Anchor_t& newAnchor) __attribute__((noinline)) {
                do {
                        newAnchor.all.low = anchor.all.low;
                        newAnchor.all.high = anchor.all.high;
                }while (newAnchor.all.low != anchor.all.low || newAnchor.all.high
                                != anchor.all.high);
        }
public:
        LockFreeDeque() {
                /* Get an SMR for memory management*/
                mm = getSMR<Node, 3> ();
                /* set initial status of anchor: "pointers null, status STABLE" */
                anchor.all.high = 0;
                anchor.all.low = 0;
        }

        void enqueue(const T &d) {
                pushRight(d);
        }

        bool dequeue(T& ret) {
                return popLeft(ret);
        }

    virtual ~LockFreeDeque() {
        Anchor_t local_anchor;
        local_anchor.all.low = anchor.all.low;
        local_anchor.all.high = anchor.all.high;

        Node * left = local_anchor.getLeft();
        Node * right = local_anchor.getRight();
        if(left!=NULL) {
            Node * current = left;
            while(current!=right) {
                Node * next = current->right.load(memory_order_relaxed);
                delete current;
                current = next;
            }
            delete right;
        }
    }

        virtual void pushRight(const T& data) __attribute__((noinline)) {
                Anchor_t old_anchor, new_anchor;
                HazardP * hp = mm->getHPRec();

#ifdef FREELIST
                Node* node = mm->newNode(hp);
                node->data = data;
#else
                /*create a new node for storing data*/
                Node* node = new Node(data);
#endif
                /* assertion for judging the last two bits is zero
                 and is free for reusing */
                assert(((unsigned long) node) % 4 == 0);

        /* loop until succeed */
                while (true) {
                        /*reading a fresh snapshot of Anchor to local */
                        copyAnchor(old_anchor);
                        /*deque is empty*/
                        if (old_anchor.fields.right == 0) {
                                assert(old_anchor.fields.left == 0);
                                /*set up a new anchor <node, node,s>*/
                                new_anchor.fields.left = (unsigned long) node;
                                new_anchor.setRight(node);
                                new_anchor.fields.status = old_anchor.fields.status;
                                /*
                                 * Insert data into to empty deque. There is no need to
                                 * stablize deque. CAS fails when others CAS changed the state
                                 * of deque, the process restarts the push attempt by reading a
                                 * fresh snapshot of Anchor
                                 */
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        return;
                                }
                        }
                        /*deque is stable*/
                        else if (old_anchor.fields.status == STABLE) {
                                node->left.store(old_anchor.getRight());
                                assert(node->left.load() != NULL);

                                /*set up a new anchor <l, node, RPUSH>*/
                                new_anchor.fields.left = old_anchor.fields.left;
                                new_anchor.setRight(node);
                                new_anchor.fields.status = RPUSH;
                                /*
                                 * Replace the anchor with new pointers, if it
                                 * succeeds, the anchor is not stable now.
                                 * CAS fails when others CAS changed the state
                                 * of deque, the process restarts the push attempt by reading a
                                 * fresh snapshot of Anchor
                                 */
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        stabilizeRight(new_anchor);
                                        return;
                                }
                        }
                        /* deque is unstable. stabilize it first. */
                        else {
                                stabilize(old_anchor);
                        }
                }
        }

        virtual void pushLeft(const T& data) __attribute__((noinline)) {
                Anchor_t old_anchor, new_anchor;
                HazardP * hp = mm->getHPRec();

#ifdef FREELIST
                Node* node = mm->newNode(hp);
                node->data = data;
#else
                /*create a new node for storing data*/
                Node* node = new Node(data);
#endif
                /* assertion for judging the last two bits is zero
                 and is free for reusing */
                assert(((unsigned long) node) % 4 == 0);

                // loop until succeed
                while (true) {
                        /*reading a fresh snapshot of Anchor to local */
                        copyAnchor(old_anchor);
                        /*deque is empty*/
                        if (old_anchor.fields.left == 0) {
                                assert(old_anchor.fields.right == 0);
                                /*set up a new anchor <node, node,s>*/
                                new_anchor.fields.left = (unsigned long) node;
                                new_anchor.setRight(node);
                                new_anchor.fields.status = old_anchor.fields.status;
                                /*
                                 * Insert data into to empty deque. There is no need to
                                 * stablize deque. CAS fails when others CAS changed the state
                                 * of deque, the process restarts the push attempt by reading a
                                 * fresh snapshot of Anchor
                                 */
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        break;
                                }
                        }
                        /*deque is stable*/
                        else if (old_anchor.fields.status == STABLE) {
                                node->right.store((Node*) old_anchor.fields.left, memory_order_relaxed);
                                assert(node->right.load() != NULL);
                                /*set up a new anchor <r, node, LPUSH>*/
                                new_anchor.fields.right = old_anchor.fields.right;
                                new_anchor.fields.left = (unsigned long) node;
                                new_anchor.fields.status = LPUSH;
                                /*
                                 * Replace the anchor with new pointers, if it
                                 * succeeds, the anchor is not stable now.
                                 * CAS fails when others CAS changed the state
                                 * of deque, the process restarts the push attempt by reading a
                                 * fresh snapshot of Anchor
                                 */
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        stabilizeLeft(new_anchor);
                                        break;
                                }
                        }
                        /* deque is unstable. stabilize it first. */
                        else {
                                stabilize(old_anchor);
                        }
                }
        }

        virtual bool popRight(T& ret) __attribute__((noinline)) {
                Anchor_t old_anchor, new_anchor;
                Node* prev;

                HazardP * hp = mm->getHPRec();

                while (true) {
                        copyAnchor(old_anchor);
                        /*deque is empty*/
                        if (old_anchor.fields.right == 0) {
                                assert(old_anchor.fields.left == 0);
                                return false;
                        }
                        /*only one element in deque*/
                        if (old_anchor.getRight() == old_anchor.getLeft()) {
                                /*create a new anchor <null,null,s>*/
                                new_anchor.fields.status = old_anchor.fields.status;
                                new_anchor.fields.left = 0;
                                new_anchor.fields.right = 0;

                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        break;
                                }
                        }
                        /*deque is stable*/
                        else if (old_anchor.fields.status == STABLE) {
                                Node * right = old_anchor.getRight();

                                mm->employ(hp, 0, (Node*) old_anchor.fields.left);
                                mm->employ(hp, 1, right);

                                assert(right != NULL);
                                if (anchor.all.high != old_anchor.all.high || anchor.all.low
                                                != old_anchor.all.low) {
                                        continue;
                                }
                                prev = right->left.load();
#ifdef DEBUG
                                if(prev==NULL) {
                                        dumpDeque(old_anchor);
                                        assert(prev!=NULL);
                                }
#endif

                                /*create a new anchor <l, prev,s>*/
                                new_anchor.fields.left = old_anchor.fields.left;
                                new_anchor.setRight(prev);
                                new_anchor.fields.status = old_anchor.fields.status;
                                /*
                                 * Replace the anchor with new value.
                                 */
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        break;
                                }
                        }
                        /*deque is unstable*/
                        else {
                                stabilize(old_anchor);
                        }
                }
                Node * right = old_anchor.getRight();
                ret = right->data;
                mm->delNode(hp, right);
                return true;
        }

#ifdef DEBUG
        void dumpDeque(Anchor_t an) {
                //print_backtrace();
                char buffer[4096];
                int pos = sprintf(buffer, "Local Anchor: left: %lx right: %lx status: %lx\n",
                                an.fields.left, an.fields.right, an.fields.status);
                pos = sprintf(buffer+pos, "Shared Anchor: left: %lx right: %lx status: %lx\n",
                                an.fields.left, an.fields.right, an.fields.status);
                cout <<buffer;
                pos =0;
                if(an.fields.left!=0) {
                        pos += sprintf(buffer+pos, "From Left:\n");
                        Node * left = (Node *) an.fields.left;
                        Node * right = an.getRight();

                        Node * current = left;
                        int i =0;
                        while(current != right) {
                                pos += sprintf(buffer+pos,
                                                "Index: %d node %p left %p right %p\n",
                                                i, current, current->left.load(), current->right.load());
                                i++;
                                current = current->right.load();
                        }
                }
                cout << buffer;
        }
#endif

        virtual bool popLeft(T& ret) __attribute__((noinline)) {
                Anchor_t old_anchor, new_anchor;
                Node* prev;
                HazardP * hp = mm->getHPRec();

                while (true) {
                        copyAnchor(old_anchor);
                        if (old_anchor.fields.right == 0) {
                                assert(old_anchor.fields.left == 0);
                                return false;
                        }

                        if (old_anchor.getRight() == old_anchor.getLeft()) {
                                new_anchor.fields.status = old_anchor.fields.status;
                                new_anchor.fields.left = 0;
                                new_anchor.fields.right = 0;
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true)
                                break;
                        } else if (old_anchor.fields.status == STABLE) {
                                mm->employ(hp, 0, (Node*) old_anchor.fields.left);
                                mm->employ(hp, 1, old_anchor.getRight());

                                //memory barrier necessary here
                                Node * left = (Node *) old_anchor.fields.left;
                                assert(left != NULL);

                                //full barrier implied
                                if (anchor.all.high != old_anchor.all.high || anchor.all.low
                                                != old_anchor.all.low) {//for memory management only
                                        continue;
                                }
                                prev = left->right.load();
#ifdef DEBUG
                                if(prev==NULL) {
                                        dumpDeque(old_anchor);
                                        assert(prev!=NULL);
                                }
#endif

                                new_anchor.fields.left = (unsigned long) prev;
                                new_anchor.fields.right = old_anchor.fields.right;
                                new_anchor.fields.status = old_anchor.fields.status;
                                if (casd(&anchor.all, &old_anchor.all, &new_anchor.all) == true) {
                                        break;
                                }
                        } else {
                                stabilize(old_anchor);
                        }
                }
                Node * left = (Node *) old_anchor.fields.left;
                ret = left->data;
                mm->delNode(hp, left);
                return true;
        }

        /*
         * @brief Get the size of the deque at this point.
         *
         * Important Node: This function is not thread-safe
         *
         * @return the number of elements in the queue
         */
        int size() {
                Anchor_t old_anchor;
                copyAnchor(old_anchor);
                if (old_anchor.fields.left == 0) {
                        return 0;
                } else if (old_anchor.getLeft() == old_anchor.getRight()) {
                        return 1;
                } else {
                        int _size = 2; 
                        Node * left = (Node *) old_anchor.fields.left;
                        Node * right = old_anchor.getRight();
                        while (true) {
                                Node * lNext = left->right.load(memory_order_relaxed);
                                if (lNext == right) {
                                        break;
                                }
                                _size++;
                                left = lNext;
                        }
                        return _size;
                }
        }

        bool empty() const {
                if (anchor.fields.left == 0) {
                        return true;
                } else {
                        return false;
                }
        }

        bool peekRight(T& ret) {
                Node* right = (Node *) (anchor.fields.right << 2);
                if (right != NULL) {
                        ret = right -> data;
                        return true;
                } else
                return false;
        }

        bool peekLeft(T& ret) {
                Node* left = (Node *) (anchor.fields.left);
                if (left != NULL) {
                        ret = left -> data;
                        return true;
                } else
                return false;
        }
private:
        void stabilize(Anchor_t a) {
                if (a.fields.status == RPUSH) {
                        stabilizeRight(a);
                } else {
                        stabilizeLeft(a);
                }
        }

        void stabilizeRight(Anchor_t a) {
                Node *prev, *prevnext;
                Anchor_t new_anchor;
                Node* al = (Node*) a.fields.left;
                Node* ar = a.getRight();
                HazardP * hp = mm->getHPRec();

                /*
                 * Prevent other threads from deletion to left and right nodes
                 */
                mm->employ(hp, 0, al);
                mm->employ(hp, 1, ar);

#ifdef DEBUG
                if(ar==NULL||ar==NULL) {
                        //this should never happen
                        dumpDeque(a);
                        assert(false);
                }
#endif
                if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                        return;
                }
                prev = ar->left.load();
#ifdef DEBUG
                if(prev==NULL) {
                        // this should never happen
                        dumpDeque(a);
                        assert(prev!=NULL);
                }
#endif

                /*
                 * prevent deletion from other threads
                 */
                mm->employ(hp, 2, prev);

                if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                        return;
                }
                prevnext = prev->right.load();

                if (prevnext != ar) {
                        if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                                return;
                        }
                        if (prev->right.compare_swap(prevnext, ar) != true) {
                                return;
                        }
                }
                new_anchor.all.high = a.all.high;
                new_anchor.all.low = a.all.low;
                new_anchor.fields.status = STABLE;

                casd(&anchor.all, &a.all, &new_anchor.all);
        }

        void stabilizeLeft(Anchor_t a) {
                Node *prev, *prevnext;
                Anchor_t new_anchor;
                Node* al = (Node*) a.fields.left;
                Node* ar = a.getRight();

                HazardP * hp = mm->getHPRec();
                /*
                 * prevent deletion from other threads
                 */
                mm->employ(hp, 0, al);
                mm->employ(hp, 1, ar);
#ifdef DEBUG
                if(ar==NULL||ar==NULL) {
                        //this should never happen
                        dumpDeque(a);
                        assert(false);
                }
#endif

                if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                        return;
                }
                prev = al->right.load();
#ifdef DEBUG
                if(prev==NULL) {
                        // this should never happen
                        dumpDeque(a);
                        assert(prev!=NULL);
                }
#endif
                /*
                 * prevent deletion from other threads
                 */
                mm->employ(hp, 2, prev);

                prevnext = prev->left.load();
                if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                        return;
                }
                if (prevnext != al) {
                        if (anchor.all.high != a.all.high || anchor.all.low != a.all.low) {
                                return;
                        }
                        if (prev->left.compare_swap(prevnext, al) != true) {
                                return;
                        }
                }
                new_anchor.all.high = a.all.high;
                new_anchor.all.low = a.all.low;
                new_anchor.fields.status = STABLE;
                casd(&anchor.all, &a.all, &new_anchor.all);
        }

};
}

#endif

---