concurrent_priority_queue.h

Go to the documentation of this file.

/*
    Copyright (c) 2005-2019 Intel 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 __TBB_concurrent_priority_queue_H
#define __TBB_concurrent_priority_queue_H

#include "atomic.h"
#include "cache_aligned_allocator.h"
#include "tbb_exception.h"
#include "tbb_stddef.h"
#include "tbb_profiling.h"
#include "internal/_aggregator_impl.h"
#include <vector>
#include <iterator>
#include <functional>
#include __TBB_STD_SWAP_HEADER

#if __TBB_INITIALIZER_LISTS_PRESENT
    #include <initializer_list>
#endif

#if __TBB_CPP11_IS_COPY_CONSTRUCTIBLE_PRESENT
    #include <type_traits>
#endif

namespace tbb {
namespace interface5 {
namespace internal {
#if __TBB_CPP11_IS_COPY_CONSTRUCTIBLE_PRESENT
    template<typename T, bool C = std::is_copy_constructible<T>::value>
    struct use_element_copy_constructor {
        typedef tbb::internal::true_type type;
    };
    template<typename T>
    struct use_element_copy_constructor <T,false> {
        typedef tbb::internal::false_type type;
    };
#else
    template<typename>
    struct use_element_copy_constructor {
        typedef tbb::internal::true_type type;
    };
#endif
} // namespace internal

using namespace tbb::internal;

template <typename T, typename Compare=std::less<T>, typename A=cache_aligned_allocator<T> >
class concurrent_priority_queue {
 public:
    typedef T value_type;

    typedef T& reference;

    typedef const T& const_reference;

    typedef size_t size_type;

    typedef ptrdiff_t difference_type;

    typedef A allocator_type;

    explicit concurrent_priority_queue(const allocator_type& a = allocator_type()) : mark(0), my_size(0), data(a)
    {
        my_aggregator.initialize_handler(my_functor_t(this));
    }

    explicit concurrent_priority_queue(size_type init_capacity, const allocator_type& a = allocator_type()) :
        mark(0), my_size(0), data(a)
    {
        data.reserve(init_capacity);
        my_aggregator.initialize_handler(my_functor_t(this));
    }

    template<typename InputIterator>
    concurrent_priority_queue(InputIterator begin, InputIterator end, const allocator_type& a = allocator_type()) :
        mark(0), data(begin, end, a)
    {
        my_aggregator.initialize_handler(my_functor_t(this));
        heapify();
        my_size = data.size();
    }

#if __TBB_INITIALIZER_LISTS_PRESENT
    concurrent_priority_queue(std::initializer_list<T> init_list, const allocator_type &a = allocator_type()) :
        mark(0),data(init_list.begin(), init_list.end(), a)
    {
        my_aggregator.initialize_handler(my_functor_t(this));
        heapify();
        my_size = data.size();
    }
#endif //# __TBB_INITIALIZER_LISTS_PRESENT


    concurrent_priority_queue(const concurrent_priority_queue& src) : mark(src.mark),
        my_size(src.my_size), data(src.data.begin(), src.data.end(), src.data.get_allocator())
    {
        my_aggregator.initialize_handler(my_functor_t(this));
        heapify();
    }


    concurrent_priority_queue(const concurrent_priority_queue& src, const allocator_type& a) : mark(src.mark),
        my_size(src.my_size), data(src.data.begin(), src.data.end(), a)
    {
        my_aggregator.initialize_handler(my_functor_t(this));
        heapify();
    }


    concurrent_priority_queue& operator=(const concurrent_priority_queue& src) {
        if (this != &src) {
            vector_t(src.data.begin(), src.data.end(), src.data.get_allocator()).swap(data);
            mark = src.mark;
            my_size = src.my_size;
        }
        return *this;
    }

#if __TBB_CPP11_RVALUE_REF_PRESENT

    concurrent_priority_queue(concurrent_priority_queue&& src) : mark(src.mark),
        my_size(src.my_size), data(std::move(src.data))
    {
        my_aggregator.initialize_handler(my_functor_t(this));
    }


    concurrent_priority_queue(concurrent_priority_queue&& src, const allocator_type& a) : mark(src.mark),
        my_size(src.my_size),
#if __TBB_ALLOCATOR_TRAITS_PRESENT
        data(std::move(src.data), a)
#else
    // Some early version of C++11 STL vector does not have a constructor of vector(vector&& , allocator).
    // It seems that the reason is absence of support of allocator_traits (stateful allocators).
        data(a)
#endif //__TBB_ALLOCATOR_TRAITS_PRESENT
    {
        my_aggregator.initialize_handler(my_functor_t(this));
#if !__TBB_ALLOCATOR_TRAITS_PRESENT
        if (a != src.data.get_allocator()){
            data.reserve(src.data.size());
            data.assign(std::make_move_iterator(src.data.begin()), std::make_move_iterator(src.data.end()));
        }else{
            data = std::move(src.data);
        }
#endif 
    }


    concurrent_priority_queue& operator=( concurrent_priority_queue&& src) {
        if (this != &src) {
            mark = src.mark;
            my_size = src.my_size;
#if !__TBB_ALLOCATOR_TRAITS_PRESENT
            if (data.get_allocator() != src.data.get_allocator()){
                vector_t(std::make_move_iterator(src.data.begin()), std::make_move_iterator(src.data.end()), data.get_allocator()).swap(data);
            }else
#endif 
            {
                data = std::move(src.data);
            }
        }
        return *this;
    }
#endif //__TBB_CPP11_RVALUE_REF_PRESENT

    template<typename InputIterator>
    void assign(InputIterator begin, InputIterator end) {
        vector_t(begin, end, data.get_allocator()).swap(data);
        mark = 0;
        my_size = data.size();
        heapify();
    }

#if __TBB_INITIALIZER_LISTS_PRESENT
    void assign(std::initializer_list<T> il) { this->assign(il.begin(), il.end()); }

    concurrent_priority_queue& operator=(std::initializer_list<T> il) {
        this->assign(il.begin(), il.end());
        return *this;
    }
#endif //# __TBB_INITIALIZER_LISTS_PRESENT


    bool empty() const { return size()==0; }


    size_type size() const { return __TBB_load_with_acquire(my_size); }


    void push(const_reference elem) {
#if __TBB_CPP11_IS_COPY_CONSTRUCTIBLE_PRESENT
        __TBB_STATIC_ASSERT( std::is_copy_constructible<value_type>::value, "The type is not copy constructible. Copying push operation is impossible." );
#endif
        cpq_operation op_data(elem, PUSH_OP);
        my_aggregator.execute(&op_data);
        if (op_data.status == FAILED) // exception thrown
            throw_exception(eid_bad_alloc);
    }

#if __TBB_CPP11_RVALUE_REF_PRESENT

    void push(value_type &&elem) {
        cpq_operation op_data(elem, PUSH_RVALUE_OP);
        my_aggregator.execute(&op_data);
        if (op_data.status == FAILED) // exception thrown
            throw_exception(eid_bad_alloc);
    }

#if __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT

    template<typename... Args>
    void emplace(Args&&... args) {
        push(value_type(std::forward<Args>(args)...));
    }
#endif /* __TBB_CPP11_VARIADIC_TEMPLATES_PRESENT */
#endif /* __TBB_CPP11_RVALUE_REF_PRESENT */


    bool try_pop(reference elem) {
        cpq_operation op_data(POP_OP);
        op_data.elem = &elem;
        my_aggregator.execute(&op_data);
        return op_data.status==SUCCEEDED;
    }


    void clear() {
        data.clear();
        mark = 0;
        my_size = 0;
    }


    void swap(concurrent_priority_queue& q) {
        using std::swap;
        data.swap(q.data);
        swap(mark, q.mark);
        swap(my_size, q.my_size);
    }

    allocator_type get_allocator() const { return data.get_allocator(); }

 private:
    enum operation_type {INVALID_OP, PUSH_OP, POP_OP, PUSH_RVALUE_OP};
    enum operation_status { WAIT=0, SUCCEEDED, FAILED };

    class cpq_operation : public aggregated_operation<cpq_operation> {
     public:
        operation_type type;
        union {
            value_type *elem;
            size_type sz;
        };
        cpq_operation(const_reference e, operation_type t) :
            type(t), elem(const_cast<value_type*>(&e)) {}
        cpq_operation(operation_type t) : type(t) {}
    };

    class my_functor_t {
        concurrent_priority_queue<T, Compare, A> *cpq;
     public:
        my_functor_t() {}
        my_functor_t(concurrent_priority_queue<T, Compare, A> *cpq_) : cpq(cpq_) {}
        void operator()(cpq_operation* op_list) {
            cpq->handle_operations(op_list);
        }
    };

    typedef tbb::internal::aggregator< my_functor_t, cpq_operation > aggregator_t;
    aggregator_t my_aggregator;
    char padding1[NFS_MaxLineSize - sizeof(aggregator_t)];
    size_type mark;
    __TBB_atomic size_type my_size;
    Compare compare;
    char padding2[NFS_MaxLineSize - (2*sizeof(size_type)) - sizeof(Compare)];

    typedef std::vector<value_type, allocator_type> vector_t;
    vector_t data;

    void handle_operations(cpq_operation *op_list) {
        cpq_operation *tmp, *pop_list=NULL;

        __TBB_ASSERT(mark == data.size(), NULL);

        // First pass processes all constant (amortized; reallocation may happen) time pushes and pops.
        while (op_list) {
            // ITT note: &(op_list->status) tag is used to cover accesses to op_list
            // node. This thread is going to handle the operation, and so will acquire it
            // and perform the associated operation w/o triggering a race condition; the
            // thread that created the operation is waiting on the status field, so when
            // this thread is done with the operation, it will perform a
            // store_with_release to give control back to the waiting thread in
            // aggregator::insert_operation.
            call_itt_notify(acquired, &(op_list->status));
            __TBB_ASSERT(op_list->type != INVALID_OP, NULL);
            tmp = op_list;
            op_list = itt_hide_load_word(op_list->next);
            if (tmp->type == POP_OP) {
                if (mark < data.size() &&
                    compare(data[0], data[data.size()-1])) {
                    // there are newly pushed elems and the last one
                    // is higher than top
                    *(tmp->elem) = tbb::internal::move(data[data.size()-1]);
                    __TBB_store_with_release(my_size, my_size-1);
                    itt_store_word_with_release(tmp->status, uintptr_t(SUCCEEDED));
                    data.pop_back();
                    __TBB_ASSERT(mark<=data.size(), NULL);
                }
                else { // no convenient item to pop; postpone
                    itt_hide_store_word(tmp->next, pop_list);
                    pop_list = tmp;
                }
            } else { // PUSH_OP or PUSH_RVALUE_OP
                __TBB_ASSERT(tmp->type == PUSH_OP || tmp->type == PUSH_RVALUE_OP, "Unknown operation" );
                __TBB_TRY{
                    if (tmp->type == PUSH_OP) {
                        push_back_helper(*(tmp->elem), typename internal::use_element_copy_constructor<value_type>::type());
                    } else {
                        data.push_back(tbb::internal::move(*(tmp->elem)));
                    }
                    __TBB_store_with_release(my_size, my_size + 1);
                    itt_store_word_with_release(tmp->status, uintptr_t(SUCCEEDED));
                } __TBB_CATCH(...) {
                    itt_store_word_with_release(tmp->status, uintptr_t(FAILED));
                }
            }
        }

        // second pass processes pop operations
        while (pop_list) {
            tmp = pop_list;
            pop_list = itt_hide_load_word(pop_list->next);
            __TBB_ASSERT(tmp->type == POP_OP, NULL);
            if (data.empty()) {
                itt_store_word_with_release(tmp->status, uintptr_t(FAILED));
            }
            else {
                __TBB_ASSERT(mark<=data.size(), NULL);
                if (mark < data.size() &&
                    compare(data[0], data[data.size()-1])) {
                    // there are newly pushed elems and the last one is
                    // higher than top
                    *(tmp->elem) = tbb::internal::move(data[data.size()-1]);
                    __TBB_store_with_release(my_size, my_size-1);
                    itt_store_word_with_release(tmp->status, uintptr_t(SUCCEEDED));
                    data.pop_back();
                }
                else { // extract top and push last element down heap
                    *(tmp->elem) = tbb::internal::move(data[0]);
                    __TBB_store_with_release(my_size, my_size-1);
                    itt_store_word_with_release(tmp->status, uintptr_t(SUCCEEDED));
                    reheap();
                }
            }
        }

        // heapify any leftover pushed elements before doing the next
        // batch of operations
        if (mark<data.size()) heapify();
        __TBB_ASSERT(mark == data.size(), NULL);
    }

    void heapify() {
        if (!mark && data.size()>0) mark = 1;
        for (; mark<data.size(); ++mark) {
            // for each unheapified element under size
            size_type cur_pos = mark;
            value_type to_place = tbb::internal::move(data[mark]);
            do { // push to_place up the heap
                size_type parent = (cur_pos-1)>>1;
                if (!compare(data[parent], to_place)) break;
                data[cur_pos] = tbb::internal::move(data[parent]);
                cur_pos = parent;
            } while( cur_pos );
            data[cur_pos] = tbb::internal::move(to_place);
        }
    }


    void reheap() {
        size_type cur_pos=0, child=1;

        while (child < mark) {
            size_type target = child;
            if (child+1 < mark && compare(data[child], data[child+1]))
                ++target;
            // target now has the higher priority child
            if (compare(data[target], data[data.size()-1])) break;
            data[cur_pos] = tbb::internal::move(data[target]);
            cur_pos = target;
            child = (cur_pos<<1)+1;
        }
        if (cur_pos != data.size()-1)
            data[cur_pos] = tbb::internal::move(data[data.size()-1]);
        data.pop_back();
        if (mark > data.size()) mark = data.size();
    }

    void push_back_helper(const T& t, tbb::internal::true_type) {
        data.push_back(t);
    }

    void push_back_helper(const T&, tbb::internal::false_type) {
        __TBB_ASSERT( false, "The type is not copy constructible. Copying push operation is impossible." );
    }
};

#if __TBB_CPP17_DEDUCTION_GUIDES_PRESENT
// Deduction guide for the constructor from two iterators
template<typename InputIterator,
         typename T = typename std::iterator_traits<InputIterator>::value_type,
         typename A = cache_aligned_allocator<T>
> concurrent_priority_queue(InputIterator, InputIterator, const A& = A())
-> concurrent_priority_queue<T, std::less<T>, A>;
#endif /* __TBB_CPP17_DEDUCTION_GUIDES_PRESENT */
} // namespace interface5

using interface5::concurrent_priority_queue;

} // namespace tbb

#endif /* __TBB_concurrent_priority_queue_H */