include/amino/queue.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 AMINO_QUEUE_H
#define AMINO_QUEUE_H

#include "smr.h"
#include <stdexcept>

namespace amino {

/*SMR is in internal namespace*/
using namespace internal;
template<typename T> class LockFreeQueue {
private:
        static const bool hasBackoff = false;

        /*
         * @brief Internal node used in the list-based queue
         */
        class Node {
        public:
                T data;
                atomic<Node*> next;
                Node() {
                        next.store(NULL, memory_order_relaxed);
                }

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

        atomic<Node *> head;

        atomic<Node *> tail;

        /*for memory management*/
        SMR<Node, 2>* mm;

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

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

public:
        LockFreeQueue() {
                /*get class's SMR*/
                mm = getSMR<Node, 2> ();

                /*dummy node. not get it from mm, for simplicity*/
                Node* dummy = new Node();
                head.store(dummy, memory_order_relaxed);
                tail.store(dummy, memory_order_relaxed);
        }

        ~LockFreeQueue() {
                Node* first = head.load(memory_order_relaxed)->next.load(
                                memory_order_relaxed);
                while (NULL != first) {
                        if (first == tail.load(memory_order_relaxed)) { /*last node in the queue*/
                                delete first;
                                break;
                        }
                        Node* next = first->next.load(memory_order_relaxed);
                        delete first;
                        first = next;
                }

                delete head.load(memory_order_relaxed);
        }

        void enqueue(T d) {
                Node *pTail; /*tail*/
                Node *pTailNext;

                int waitForBackoff = 10000;
#ifdef FREELIST
        Node * node = mm->newNode();
        node->data = d;
#else
                Node * node = new Node(d);
#endif

                while (true) {
                        /* used as first parameter in compare_and_swap. compare_and_swap
                         * cannot accept NULL as its first parameter
                         */
                        Node* temp = NULL;
                        pTail = tail.load(memory_order_relaxed);
                        /*hazard pointer, protected by SMR */
                        mm->employ(0, pTail);

                        /*full barrier necessary*/
                        if (tail.load(memory_order_seq_cst) != pTail)
                                continue;
                        pTailNext = pTail->next.load(memory_order_acquire);
                        /*load-load barrier required*/

                        /*are latest tail and t_next consistent?*/
                        if (tail.load(memory_order_relaxed) != pTail)
                                continue;

                        /*is tail pointing to the last node?*/
                        if (pTailNext != NULL) {
                                /*try to swing tail to the next node*/
                                tail.compare_swap(pTail, pTailNext);
                                continue;
                        }

                        /*enqueue by CAS tail pointer*/
                        if (pTail->next.compare_swap(temp, node))
                                break; /*enqueue is done*/

                        if (hasBackoff) {
                                usleep(waitForBackoff);
                                waitForBackoff <<= 1;
                        }
                }
                /*try to swing tail to the inserted node*/
                tail.compare_swap(pTail, node);
        }

        bool dequeue(T& ret) {
                Node *pHead;
                Node *pTail;
                Node *pHeadNext;

                int waitForBackoff = 10000;
                while (true) {
                        pHead = head.load(memory_order_relaxed);
                        /*hazard pointer, protected by SMR */
                        mm->employ(0, pHead);
                        /*full barrier , necessary*/
                        if (head.load(memory_order_seq_cst) != pHead)
                                continue;

                        pTail = tail.load(memory_order_relaxed);
                        pHeadNext = pHead->next.load(memory_order_relaxed);
                        /*hazard pointer, protected by SMR */
                        mm->employ(1, pHeadNext);
                        /*full barrier, necessary*/
                        /*are head, tail, and h_next consistent?*/
                        if (head.load(memory_order_seq_cst) != pHead)
                                continue;
                        if (pHeadNext == NULL) /*empty queue*/
                                return false;
                        if (pHead == pTail) { /*is tail falling behind*/
                                tail.compare_swap(pTail, pHeadNext); /*try to advance pTail*/
                                continue;
                        }
                        ret = pHeadNext->data;

                        /*barrier*/

                        if (head.compare_swap(pHead, pHeadNext)) /*try to swing head to the next node*/
                                break;

                        if (hasBackoff) {
                                usleep(waitForBackoff);
                                waitForBackoff <<= 1;
                        }
                }
                mm->delNode(pHead); /*free node from SMR*/
                return true;
        }

        bool empty() {
                return NULL == head.load(memory_order_relaxed)->next.load(
                                memory_order_relaxed);
        }

        int size() {
                int result = 0;
                Node *front = head.load(memory_order_relaxed)->next.load(memory_order_relaxed);
                while (front != NULL) {
                        ++result;
                        if (front == tail.load(memory_order_relaxed))
                                break;
                        front = front->next.load(memory_order_relaxed);
                }
                return result;
        }

        bool peekFront(T& top) {
                Node *front = (head.load(memory_order_relaxed)->next).load(memory_order_relaxed);
                if (front == NULL) {
                        return false;
                }
                top = front->data;
                return true;
        }
};
}//namespace amino

#endif

---