include/amino/lockfree_priorityqueue.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.
 *
 */
/*
 * lockfree_priorityqueue.h
 *
 *  Created on: Sep 11, 2008
 *      Author: daixj
 */

#ifndef LOCKFREE_PRIORITYQUEUE_H_
#define LOCKFREE_PRIORITYQUEUE_H_

#include <amino/smr.h>

#include <vector>
#include <cstdlib>
#include <ctime>
#include <iostream>

namespace amino {
using namespace internal;

#define COPY_NODE(node) (node)
#define READ_NODE(node) (IS_MARKED(node) ? NULL : (node))
#define RELEASE_NODE(node) (node)
#define GET_UNMARKED(p) ((PQNode<E>*)(((long)(p))&(~3)))
#define GET_UNMARKED_VALUE(p) ((Value<E>*)(((long)(p))&(~3)))
/* set the last bit of pointers */
#define GET_MARKED(p) ((PQNode<E>*)(((long)(p))|(1)))
/* set the last bit of pointers */
#define GET_MARKED_VALUE(p) ((Value<E>*)(((long)(p))|(1)))
/* get the last bit of pointers */
#define IS_MARKED(p) (((long)(p))&(1))

#define MAXLEVEL 10
#define SLCONST 0.5

template<typename E> struct Value {
        E v;

        Value() {
        }

        Value(const E& value) :
                v(value) {
        }
};

template<typename E> struct PQNode {
        int level;

        /* as the nodes, which are used in the lock-free memory management
         * scheme, will be reused for the same purpose when re-allocated
         * again after being reclaimed, the individual fields of the nodes
         * that are not being reclaimed, the individual fields of the nodes
         * that are not part of the memory management scheme will be intact.
         * the validLevel field can therefore be used for indicating if the
         * current node can be used for possibly traversing further on a
         * certain level. A value of 0 indicates that this node can not be
         * used for traversing at all, as it is possibly reclaimed or not yet
         * inserted. As the validLevel field is only set to 0 directly before
         * reclamation, a positive value indicates that the node is allocated.
         * A value of n+1 indicated that this node has been inserted up to level n.
         */
        int validLevel;
        atomic<Value<E>*> value;
        PQNode<E>* prev;
        atomic<PQNode<E>*> next[MAXLEVEL];
        E key;

        PQNode() :
                level(0), validLevel(-1), prev(NULL),key() {
                value.store(NULL);
                for (int i = 0; i < MAXLEVEL; ++i) {
                        next[i].store(NULL, memory_order_relaxed);
                }
        }

        PQNode(const E& v) :
                prev(NULL), validLevel(-1), level(0), key(v) {
                value.store(new Value<E> (v), memory_order_relaxed);
                for (int i = 0; i < MAXLEVEL; ++i) {
                        next[i].store(NULL, memory_order_relaxed);
                }
        }

        PQNode(int l, const E& v) :
                level(l), validLevel(-1), prev(NULL), key(v) {
                value.store(new Value<E> (v), memory_order_relaxed);
                for (int i = 0; i < MAXLEVEL; ++i) {
                        next[i].store(NULL, memory_order_relaxed);
                }
        }

        ~PQNode() {
                delete GET_UNMARKED_VALUE(value.load(memory_order_relaxed));
        }
};

template<typename E> class LockFreePriorityQueue {
        /*head pointer*/
        PQNode<E>* head;
        /*tail pointer*/
        PQNode<E>* tail;

        PQNode<E>* INVALID;

        SMR<PQNode<E>, MAXLEVEL>* mm; // MAXLEVEL for next pointer on every level
        typedef typename SMR<PQNode<E>, MAXLEVEL>::HP_Rec HazardP;

public:
        LockFreePriorityQueue() {
                mm = getSMR<PQNode<E>, MAXLEVEL> ();
                head = new PQNode<E> ();
                head->validLevel = MAXLEVEL - 1;
                tail = new PQNode<E> ();
                tail->validLevel = MAXLEVEL - 1;
                INVALID = new PQNode<E> ();

                for (int i = 0; i < MAXLEVEL; ++i) {
                        head->next[i].store(tail, memory_order_relaxed);
                }

                srand(time(NULL));
        }

        ~LockFreePriorityQueue() {
                PQNode<E> *curr = head->next[0].load(memory_order_relaxed);
        PQNode<E> *tmp = NULL;
                while (tail != curr) {
            cout << "destructing node .........." << endl;
            tmp = curr;
            curr = curr->next[0].load(memory_order_relaxed);
            delete tmp;
                }

                delete head;
                delete tail;
                delete INVALID;
        
        }

        bool empty() {
                return head->next[0].load(memory_order_relaxed) == tail;
        }

        int size() {
                int size = 0;
                for (PQNode<E>* itr = head->next[0].load(memory_order_relaxed); itr
                                != tail; itr = itr->next[0].load(memory_order_relaxed)) {
                        ++size;
                }
                return size;
        }

        bool peek(E& top) {
                top = head->next[0].load(memory_order_relaxed)->value.load(
                                memory_order_relaxed)->v;
                if (top == tail) {
                        return false;
                } else {
                        return true;
                }
        }

        bool enqueue(const E& value) {
                return insert(value);
        }

        bool dequeue(E& result) {
                return deleteMin(result);
        }

        bool insert(const E& value) {
//        cout << "inserting............... " << value << endl;
                E key = value;
                int curLevel = randomLevel();
                PQNode<E>* savedNodes[curLevel];
        HazardP * hp = mm->getHPRec();
#ifdef FREELIST
        PQNode<E>* newNode = mm->newNode(hp); /*Allocate a new node*/
        newNode->level = curLevel;
        newNode->key = value;
        newNode->value.store(new Value<E> (value), memory_order_relaxed);
#else
        PQNode<E>* newNode = new PQNode<E> (curLevel, value);/*Allocate a new node*/
#endif
                //              COPY_NODE(newNode);
                //              newNode is local, doesn't need to protected by SMR

try_again:
                PQNode<E>* node1 = COPY_NODE(head);
                mm->employ(hp,0,head);
                if(node1 != head) {
                        goto try_again;
                }
                
                PQNode<E>* node2;
                Value<E>* value2;
                /*
                 * search phase to find the node after which the new node(newNode) should
                 * be inserted. This search phase starts from the head node at highest
                 * level and travels down to the lowest level until the correct node
                 * is found(node1). when going down one level, the last node traversed
                 * on that level is remembered(savedNodes) for later use (this is where
                 * we should insert the new node at that level).
                 */
                for (int i = MAXLEVEL - 1; i >= 1; --i) {
                        node2 = scanKey(node1, i, key);
//                      RELEASE_NODE(node2);
            mm->retire(hp,node2);
                        if (i < curLevel) {
                                savedNodes[i] = COPY_NODE(node1);
                
                                mm->employ(hp,i,node1);
                                if(node1 != savedNodes[i]) {
                                        goto try_again;
                }
                        }
                }

                while (true) {
                        node2 = scanKey(node1, 0, key);
                        value2 = node2->value.load(memory_order_relaxed);
                        /*
                         * now it is possible that there already exists a node with
                         * the same priority as of the new node, the value of the old node(node2)
                         * is changed atomically with a CAS.
                         */
                        if ((node2 != tail) && (!IS_MARKED(value2) && (node2->key == key))) {
                                if (node2->value.compare_swap(value2, new Value<E> (value))) {
                                        cout << "found same value: " << value2->v << ":" << value
                                                        << endl;
                                        // TODO memory leak here?
//                                      RELEASE_NODE(node1);
//                                      RELEASE_NODE(node2);
                    mm->retire(hp,node1);
                    mm->retire(hp,node2);
                                        for (int i = 1; i < curLevel; ++i) {
//                                              RELEASE_NODE(savedNodes[i]);
                        mm->retire(hp,savedNodes[i]);
                                        }
//                                      RELEASE_NODE(newNode);
//                                      RELEASE_NODE(newNode);
                                        delete value2;
                    mm->delNode(hp,newNode);
                    cout << "found same value" << endl;
                                        return true;
                                } else {
//                                      RELEASE_NODE(node2);
                    mm->retire(hp,node2);
                                        continue;
                                }
                        }
                        /*
                         * otherwise, it starts trying to insert the new node starting with
                         * the lowest level and increasing up to the level of the new node.
                         * the next pointers of the nodes (to become previous) are changed
                         * atomically with a CAS.
                         */
                        newNode->next[0].store(node2, memory_order_relaxed);
//                      RELEASE_NODE(node2);
            mm->retire(hp,node2);
                        if (node1->next[0].compare_swap(node2, newNode)) {
                                //                              cout << "CAS succeed\n";
//                              RELEASE_NODE(node1);
                mm->retire(hp,node1);
                                break;
                        }
                        //TODO add backoff
                }
                for (int i = 1; i < curLevel; ++i) {
                        newNode->validLevel = i;
                        node1 = savedNodes[i];
                        while (true) {
                                node2 = scanKey(node1, i, key);
                                newNode->next[i].store(node2);
//                              RELEASE_NODE(node2);
                mm->retire(hp,node2);
                                /*
                                 * After the new node has been inserted at the lowest level,
                                 * it is possible that it is deleted by a concurrent deleteMin
                                 * operation before it has been inserted at all levels.
                                 */
                                if (IS_MARKED(newNode->value.load(memory_order_relaxed))
                                || node1->next[i].compare_swap(node2, newNode)) {
//                                      RELEASE_NODE(node1);
                    mm->retire(hp,node1);
                                        break;
                                }
                                //TODO add backoff
                        }
                }

                newNode->validLevel = curLevel;
                if (IS_MARKED(newNode->value.load(memory_order_relaxed))) {
                        newNode = helpDelete(newNode, 0);
                }
//              RELEASE_NODE(newNode);
        // newNode is local variable, no need to release
                return true;
        }

        bool deleteMin(E& result) {
                PQNode<E>* node1;
                PQNode<E>* node2;
                PQNode<E>* last;
                Value<E>* value;
        
        HazardP * hp = mm->getHPRec();
                /*
                 * start from the head node and find the first node (node1) in the list
                 * that does not have its deletion mark on the value set
                 */
        try_again:
        PQNode<E>* prev = COPY_NODE(head);
                mm->employ(hp,0,head);
                if(prev != head) {
                        goto try_again;
                }

                while (true) {
                        node1 = readNext(prev, 0);
                        if (node1 == tail) {
//                              RELEASE_NODE(prev);
//                              RELEASE_NODE(node1);
                mm->retire(hp,prev);
                mm->retire(hp,node1);
                                return false;
                        }
                        retry: value = node1->value.load(memory_order_relaxed);
            result = value->v;
                        if (!IS_MARKED(value)) {
                                /*
                                 * it tries to set this deletion mark using CAS primitive.
                                 * If it succeeds it also writes a valid pointer to the prev
                                 * field of the node. this prev field is necessary in order to
                                 * increase the performance of concurrent helpDelete operation,
                                 * these operations otherwise would have to search for the previous
                                 * node in order to complete the deletion.
                                 */
                                if (node1->value.compare_swap(value, GET_MARKED_VALUE(value))) {
                                        node1->prev = prev;
                                        break;
                                } else {
                                        goto retry;
                                }
                        } else if (IS_MARKED(value)) {
                                node1 = helpDelete(node1, 0);
                        }
//                      RELEASE_NODE(prev);
            mm->retire(hp,prev);
                        prev = node1;
                }
                /*
                 * the next step is to mark the deletion bits of the next pointer
                 * in the nodes, starting with the lowest level and going upwards,
                 * using the CAS primitive in each step
                 */
                assert(node1 != tail && node1 != head);
                for (int i = 0; i < node1->level; ++i) {
                        do {
                                node2 = node1->next[i].load(memory_order_relaxed);
                        } while (!IS_MARKED(node2) && !node1->next[i].compare_swap(node2, GET_MARKED(node2)));
                }
                prev = COPY_NODE(head);
                mm->employ(hp,0,head);
                if(prev != head) {
                        goto try_again;
                }
                /*
                 * starts the actual deletion by changing the next pointers of the
                 * previous node(prev), starting at the highest level and continuing
                 * downwards. the reason for doing the deletion in decreasing order
                 * of levels, is that concurrent search operations also start at the
                 * highest level and processed downwards, in this way the concurrent search
                 * operations will sooner avoid traversing this node. the procedure performed
                 * by the deletemin operation in order to change each next pointer of
                 * the previous node, is to first search for the previous node and then
                 * perform the CAS primitive until it succeeds.
                 */
                for (int i = node1->level - 1; i >= 0; --i) {
                        while (true) {
                                assert(node1 != tail && node1 != head);
                                if (GET_UNMARKED(node1->next[i].load(memory_order_relaxed)) == INVALID) {
                                        break;
                                }
                                last = scanKey(prev, i, node1->key);
//                              RELEASE_NODE(last);
                mm->retire(hp,last);
                                if (last != node1 || GET_UNMARKED(node1->next[i].load(memory_order_relaxed)) == INVALID) {
                                        break;
                                }
                                if (prev->next[i].compare_swap(node1, GET_UNMARKED(
                                                node1->next[i].load(memory_order_relaxed)))) {
                                        node1->next[i].store(INVALID, memory_order_relaxed);
                                        break;
                                }
                                if (GET_UNMARKED(node1->next[i].load(memory_order_relaxed)) == INVALID) {
                                        break;
                                }
                                //TODO add backoff
                        }
                }
//              RELEASE_NODE(prev);
        mm->retire(hp,prev);
//              RELEASE_NODE(node1);
//              RELEASE_NODE(node1); /* delete the node */
//                 cout << "delNode ........................ " << node1->key << ":" << GET_UNMARKED_VALUE(node1->value.load(memory_order_relaxed))->v<< endl;
        mm->delNode(node1);
                return true;
        }

        void dumpQueue() {
                cout << "dumping queue...................." << endl;
                cout << "head:" << head << "->";
                for (PQNode<E>* itr = head->next[0].load(memory_order_relaxed); itr
                                != tail; itr = itr->next[0].load(memory_order_relaxed)) {
                        cout << GET_UNMARKED(itr) << "->";
                }
                cout << tail << ":tail" << endl;
        }

private:
        PQNode<E>* readNext(PQNode<E>*& node, int curLevel) {
                //              cout<< "readNext" << endl;
                PQNode<E>* nextNode;
                PQNode<E>* tmpNode;

                assert(node != tail);
                if (IS_MARKED(node->value.load(memory_order_relaxed))) {
                        node = helpDelete(node, curLevel);
                }
                tmpNode = node->next[curLevel].load(memory_order_relaxed);
                nextNode = READ_NODE(tmpNode);
                while (NULL == nextNode) {
                        node = helpDelete(node, curLevel);
                        tmpNode = node->next[curLevel].load(memory_order_relaxed);
                        nextNode = READ_NODE(tmpNode);
                }
                assert(!IS_MARKED(nextNode));
                return nextNode;
        }

        PQNode<E>* scanKey(PQNode<E>*& node, int curLevel, const E& expectedKey) {
        HazardP * hp = mm->getHPRec();
                //              cout<< "scanKey" << endl;
                PQNode<E>* nextNode = readNext(node, curLevel);
                //              Value<E>* nextValue = nextNode->value.load(memory_order_relaxed);
                while ((nextNode != tail) && ((nextNode == head) || nextNode->key
                                < expectedKey)) {
//                      RELEASE_NODE(node);
            mm->retire(hp,node);
                        //                      cout << nextNode->value.load(memory_order_relaxed)->v << ", ";
                        node = nextNode;
                        nextNode = readNext(node, curLevel);
                }
                //              cout<<endl;
                return nextNode;
        }

        /*
         *it tries to fulfill the deletion on the current level and returns when it is completed.
         */
        PQNode<E>* helpDelete(PQNode<E>* node, int curLevel) {
                //              cout<< "helpDelete" << endl;
                //              dumpQueue();
        HazardP * hp = mm->getHPRec();
        
                PQNode<E>* node2;
                PQNode<E>* last;
                /*
                 * it starts with setting the deletion mark on all next pointers
                 * in case they have not been set.
                 */
                for (int i = curLevel; i < node->level; ++i) {
                        do {
                                node2 = node->next[i].load(memory_order_relaxed);
                        } while (!IS_MARKED(node2) && !node->next[i].compare_swap(node2, GET_MARKED(
                                        node2)));
                }
                /*
                 * it checks if the node given in the prev field is valid for deletion of
                 * on the current level
                 */
                PQNode<E>* prev = node->prev;
                if (NULL == prev || (curLevel >= prev->validLevel)) {
                        /*
                         * otherwise it searches for the correct node (prev)
                         */
                try_again:
            prev = COPY_NODE(head);
            mm->employ(hp,0,head);
            if(prev != head) {
                goto try_again;
            }
            
                        for (int i = MAXLEVEL - 1; i >= curLevel; --i) {
                                node2 = scanKey(prev, i, node->key);
//                              RELEASE_NODE(node2);
                mm->retire(hp, node2);
                        }
                } else {
                        //COPY_NODE(prev); // prev is local variable, no need to copy_node
                }
                /* the actual deletion of this node on the current level takes place.
                 * this operation might execute concurrently with the corresponding
                 * deleteMin operation, and therefore both operations synchronize with
                 * each other in order to avoid executing sub-operations that have already
                 * been performed.
                 */
                while (true) {
                        assert(node != tail && node != head);
                        if (GET_UNMARKED(node->next[curLevel].load(memory_order_relaxed)) == INVALID) {
                                break;
                        }
                        last = scanKey(prev, curLevel, node->key);
//                      RELEASE_NODE(last);
            mm->retire(hp,last);
                        if (last != node || GET_UNMARKED(node->next[curLevel].load(memory_order_relaxed)) == INVALID) {
                                break;
                        }
                        if (prev->next[curLevel].compare_swap(node, GET_UNMARKED(
                                        node->next[curLevel].load(memory_order_relaxed)))) {
                                node->next[curLevel].store(INVALID);
                                break;
                        }
                        if (GET_UNMARKED(node->next[curLevel].load(memory_order_relaxed)) == INVALID) {
                                break;
                        }
                        //TODO add backoff
                }
//              RELEASE_NODE(node);
        mm->retire(hp,node);
                return prev;
        }

        int randomLevel() {
                int v = 1;
                while (((static_cast<double> (rand()) / RAND_MAX) < SLCONST) && (v
                                < MAXLEVEL - 1)) {
                        ++v;
                }
                //              cout << v << " ";
                return v;
        }

};

}

#endif /* LOCKFREE_PRIORITYQUEUE_H_ */

---