// Copyright (C) 2014 Ian Forbed // Copyright (C) 2014 Vicente J. Botet Escriba // // Distributed under the Boost Software License, Version 1.0. (See accompanying // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) // #ifndef BOOST_THREAD_SYNC_PRIORITY_QUEUE #define BOOST_THREAD_SYNC_PRIORITY_QUEUE #include #include #include #include #include #include #include #include #include #include #include #include #include #include namespace boost { namespace detail { template < class Type, class Container = csbl::vector, class Compare = std::less > class priority_queue { private: Container _elements; Compare _compare; public: typedef Type value_type; typedef typename Container::size_type size_type; explicit priority_queue(const Compare& compare = Compare()) : _elements(), _compare(compare) { } size_type size() const { return _elements.size(); } bool empty() const { return _elements.empty(); } void push(Type const& element) { _elements.push_back(element); std::push_heap(_elements.begin(), _elements.end(), _compare); } void push(BOOST_RV_REF(Type) element) { _elements.push_back(boost::move(element)); std::push_heap(_elements.begin(), _elements.end(), _compare); } void pop() { std::pop_heap(_elements.begin(), _elements.end(), _compare); _elements.pop_back(); } Type pull() { Type result = boost::move(_elements.front()); pop(); return boost::move(result); } Type const& top() { return _elements.front(); } }; } namespace concurrent { template , class Compare = std::less > class sync_priority_queue : public detail::sync_queue_base > { typedef detail::sync_queue_base > super; public: typedef ValueType value_type; //typedef typename super::value_type value_type; // fixme typedef typename super::underlying_queue_type underlying_queue_type; typedef typename super::size_type size_type; typedef typename super::op_status op_status; typedef chrono::steady_clock clock; protected: public: sync_priority_queue() {} ~sync_priority_queue() { if(!super::closed()) { super::close(); } } void push(const ValueType& elem); void push(BOOST_THREAD_RV_REF(ValueType) elem); queue_op_status try_push(const ValueType& elem); queue_op_status try_push(BOOST_THREAD_RV_REF(ValueType) elem); ValueType pull(); void pull(ValueType&); queue_op_status pull_until(const clock::time_point&, ValueType&); queue_op_status pull_for(const clock::duration&, ValueType&); queue_op_status try_pull(ValueType& elem); queue_op_status wait_pull(ValueType& elem); queue_op_status nonblocking_pull(ValueType&); private: void push(unique_lock&, const ValueType& elem); void push(lock_guard&, const ValueType& elem); void push(unique_lock&, BOOST_THREAD_RV_REF(ValueType) elem); void push(lock_guard&, BOOST_THREAD_RV_REF(ValueType) elem); queue_op_status try_push(unique_lock&, const ValueType& elem); queue_op_status try_push(unique_lock&, BOOST_THREAD_RV_REF(ValueType) elem); ValueType pull(unique_lock&); ValueType pull(lock_guard&); void pull(unique_lock&, ValueType&); void pull(lock_guard&, ValueType&); queue_op_status try_pull(lock_guard& lk, ValueType& elem); queue_op_status try_pull(unique_lock& lk, ValueType& elem); queue_op_status wait_pull(unique_lock& lk, ValueType& elem); queue_op_status nonblocking_pull(unique_lock& lk, ValueType&); sync_priority_queue(const sync_priority_queue&); sync_priority_queue& operator= (const sync_priority_queue&); sync_priority_queue(BOOST_THREAD_RV_REF(sync_priority_queue)); sync_priority_queue& operator= (BOOST_THREAD_RV_REF(sync_priority_queue)); }; //end class ////////////////////// template void sync_priority_queue::push(unique_lock& lk, const T& elem) { super::throw_if_closed(lk); super::data_.push(elem); super::notify_not_empty_if_needed(lk); } template void sync_priority_queue::push(lock_guard& lk, const T& elem) { super::throw_if_closed(lk); super::data_.push(elem); super::notify_not_empty_if_needed(lk); } template void sync_priority_queue::push(const T& elem) { lock_guard lk(super::mtx_); push(lk, elem); } ////////////////////// template void sync_priority_queue::push(unique_lock& lk, BOOST_THREAD_RV_REF(T) elem) { super::throw_if_closed(lk); super::data_.push(boost::move(elem)); super::notify_not_empty_if_needed(lk); } template void sync_priority_queue::push(lock_guard& lk, BOOST_THREAD_RV_REF(T) elem) { super::throw_if_closed(lk); super::data_.push(boost::move(elem)); super::notify_not_empty_if_needed(lk); } template void sync_priority_queue::push(BOOST_THREAD_RV_REF(T) elem) { lock_guard lk(super::mtx_); push(lk, boost::move(elem)); } ////////////////////// template queue_op_status sync_priority_queue::try_push(const T& elem) { lock_guard lk(super::mtx_); if (super::closed(lk)) return queue_op_status::closed; push(lk, elem); return queue_op_status::success; } ////////////////////// template queue_op_status sync_priority_queue::try_push(BOOST_THREAD_RV_REF(T) elem) { lock_guard lk(super::mtx_); if (super::closed(lk)) return queue_op_status::closed; push(lk, boost::move(elem)); return queue_op_status::success; } ////////////////////// template T sync_priority_queue::pull(unique_lock&) { return super::data_.pull(); } template T sync_priority_queue::pull(lock_guard&) { return super::data_.pull(); } template T sync_priority_queue::pull() { unique_lock lk(super::mtx_); super::wait_until_not_empty(lk); return pull(lk); } ////////////////////// template void sync_priority_queue::pull(unique_lock&, T& elem) { elem = super::data_.pull(); } template void sync_priority_queue::pull(lock_guard&, T& elem) { elem = super::data_.pull(); } template void sync_priority_queue::pull(T& elem) { unique_lock lk(super::mtx_); super::wait_until_not_empty(lk); pull(lk, elem); } ////////////////////// template queue_op_status sync_priority_queue::pull_until(const clock::time_point& tp, T& elem) { unique_lock lk(super::mtx_); if (queue_op_status::timeout == super::wait_until_not_empty_until(lk, tp)) return queue_op_status::timeout; pull(lk, elem); return queue_op_status::success; } ////////////////////// template queue_op_status sync_priority_queue::pull_for(const clock::duration& dura, T& elem) { return pull_until(clock::now() + dura, elem); } ////////////////////// template queue_op_status sync_priority_queue::try_pull(unique_lock& lk, T& elem) { if (super::empty(lk)) { if (super::closed(lk)) return queue_op_status::closed; return queue_op_status::empty; } pull(lk, elem); return queue_op_status::success; } template queue_op_status sync_priority_queue::try_pull(lock_guard& lk, T& elem) { if (super::empty(lk)) { if (super::closed(lk)) return queue_op_status::closed; return queue_op_status::empty; } pull(lk, elem); return queue_op_status::success; } template queue_op_status sync_priority_queue::try_pull(T& elem) { lock_guard lk(super::mtx_); return try_pull(lk, elem); } ////////////////////// template queue_op_status sync_priority_queue::wait_pull(unique_lock& lk, T& elem) { if (super::empty(lk)) { if (super::closed(lk)) return queue_op_status::closed; } bool has_been_closed = super::wait_until_not_empty_or_closed(lk); if (has_been_closed) return queue_op_status::closed; pull(lk, elem); return queue_op_status::success; } template queue_op_status sync_priority_queue::wait_pull(T& elem) { unique_lock lk(super::mtx_); return wait_pull(lk, elem); } ////////////////////// template queue_op_status sync_priority_queue::nonblocking_pull(T& elem) { unique_lock lk(super::mtx_, try_to_lock); if (!lk.owns_lock()) return queue_op_status::busy; return try_pull(lk, elem); } } //end concurrent namespace using concurrent::sync_priority_queue; } //end boost namespace #include #endif