SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
- OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef CDSLIB_INTRUSIVE_MSPRIORITY_QUEUE_H
struct stat {
typedef Counter event_counter ; ///< Event counter type
- event_counter m_nPushCount ; ///< Count of success push operation
- event_counter m_nPopCount ; ///< Count of success pop operation
- event_counter m_nPushFailCount ; ///< Count of failed ("the queue is full") push operation
- event_counter m_nPopFailCount ; ///< Count of failed ("the queue is empty") pop operation
- event_counter m_nPushHeapifySwapCount ; ///< Count of item swapping when heapifying in push
- event_counter m_nPopHeapifySwapCount ; ///< Count of item swapping when heapifying in pop
+ event_counter m_nPushCount; ///< Count of success push operation
+ event_counter m_nPopCount; ///< Count of success pop operation
+ event_counter m_nPushFailCount; ///< Count of failed ("the queue is full") push operation
+ event_counter m_nPopFailCount; ///< Count of failed ("the queue is empty") pop operation
+ event_counter m_nPushHeapifySwapCount; ///< Count of item swapping when heapifying in push
+ event_counter m_nPopHeapifySwapCount; ///< Count of item swapping when heapifying in pop
+ event_counter m_nItemMovedTop; ///< Count of events when \p push() encountered that inserted item was moved to top by a concurrent \p pop()
+ event_counter m_nItemMovedUp; ///< Count of events when \p push() encountered that inserted item was moved upwards by a concurrent \p pop()
+ event_counter m_nPushEmptyPass; ///< Count of empty pass during heapify via concurrent operations
//@cond
void onPushSuccess() { ++m_nPushCount ;}
void onPopFailed() { ++m_nPopFailCount ;}
void onPushHeapifySwap() { ++m_nPushHeapifySwapCount ;}
void onPopHeapifySwap() { ++m_nPopHeapifySwapCount ;}
+
+ void onItemMovedTop() { ++m_nItemMovedTop ;}
+ void onItemMovedUp() { ++m_nItemMovedUp ;}
+ void onPushEmptyPass() { ++m_nPushEmptyPass ;}
//@endcond
};
/// MSPriorityQueue empty statistics
struct empty_stat {
//@cond
- void onPushSuccess() {}
- void onPopSuccess() {}
- void onPushFailed() {}
- void onPopFailed() {}
- void onPushHeapifySwap() {}
- void onPopHeapifySwap() {}
+ void onPushSuccess() const {}
+ void onPopSuccess() const {}
+ void onPushFailed() const {}
+ void onPopFailed() const {}
+ void onPushHeapifySwap() const {}
+ void onPopHeapifySwap() const {}
+
+ void onItemMovedTop() const {}
+ void onItemMovedUp() const {}
+ void onPushEmptyPass() const {}
//@endcond
};
+ class monotonic_counter
+ {
+ public:
+ typedef size_t counter_type;
+
+ monotonic_counter()
+ : m_nCounter(0)
+ {}
+
+ size_t inc()
+ {
+ return ++m_nCounter;
+ }
+
+ size_t dec()
+ {
+ return m_nCounter--;
+ }
+
+ size_t value() const
+ {
+ return m_nCounter;
+ }
+
+ private:
+ size_t m_nCounter;
+ };
+
/// MSPriorityQueue traits
struct traits {
/// Storage type
/**
- The storage type for the heap array. Default is \p cds::opt::v::dynamic_buffer.
+ The storage type for the heap array. Default is \p cds::opt::v::initialized_dynamic_buffer.
You may specify any type of buffer's value since at instantiation time
the \p buffer::rebind member metafunction is called to change type
of values stored in the buffer.
*/
- typedef opt::v::dynamic_buffer<void *> buffer;
+ typedef opt::v::initialized_dynamic_buffer<void *> buffer;
/// Priority compare functor
/**
*/
typedef opt::none less;
- /// Type of mutual-exclusion lock
+ /// Type of mutual-exclusion lock. The lock is not need to be recursive.
typedef cds::sync::spin lock_type;
/// Back-off strategy
or any other with interface like \p %mspriority_queue::stat
*/
typedef empty_stat stat;
+
+ /// Item counter type
+ typedef cds::bitop::bit_reverse_counter<> item_counter;
+
+ /// Fairness
+ static bool const fairness = true;
};
/// Metafunction converting option list to traits
/**
\p Options:
- - \p opt::buffer - the buffer type for heap array. Possible type are: \p opt::v::static_buffer, \p opt::v::dynamic_buffer.
- Default is \p %opt::v::dynamic_buffer.
- You may specify any type of values for the buffer since at instantiation time
+ - \p opt::buffer - the buffer type for heap array. Possible type are: \p opt::v::initialized_static_buffer, \p opt::v::initialized_dynamic_buffer.
+ Default is \p %opt::v::initialized_dynamic_buffer.
+ You may specify any type of value for the buffer since at instantiation time
the \p buffer::rebind member metafunction is called to change the type of values stored in the buffer.
- \p opt::compare - priority compare functor. No default functor is provided.
If the option is not specified, the \p opt::less is used.
typedef typename opt::details::make_comparator< value_type, traits >::type key_comparator;
# endif
- typedef typename traits::lock_type lock_type ; ///< heap's size lock type
- typedef typename traits::back_off back_off ; ///< Back-off strategy
- typedef typename traits::stat stat ; ///< internal statistics type
+ typedef typename traits::lock_type lock_type; ///< heap's size lock type
+ typedef typename traits::back_off back_off; ///< Back-off strategy
+ typedef typename traits::stat stat; ///< internal statistics type
protected:
//@cond
typedef typename traits::buffer::template rebind<node>::other buffer_type ; ///< Heap array buffer type
//@cond
- typedef cds::bitop::bit_reverse_counter<> item_counter_type;
+ typedef typename traits::item_counter item_counter_type;
typedef typename item_counter_type::counter_type counter_type;
//@endcond
+ static const bool c_bFairQueue = traits::fairness;
+
protected:
item_counter_type m_ItemCounter ; ///< Item counter
mutable lock_type m_Lock ; ///< Heap's size lock
public:
/// Constructs empty priority queue
/**
- For \p cds::opt::v::static_buffer the \p nCapacity parameter is ignored.
+ For \p cds::opt::v::initialized_static_buffer the \p nCapacity parameter is ignored.
*/
MSPriorityQueue( size_t nCapacity )
: m_Heap( nCapacity )
node& refNode = m_Heap[i];
refNode.lock();
m_Lock.unlock();
+ assert( refNode.m_nTag == tag_type( Empty ));
+ assert( refNode.m_pVal == nullptr );
refNode.m_pVal = &val;
refNode.m_nTag = curId;
refNode.unlock();
- // Move item towards top of the heap while it has higher priority than parent
+ // Move item towards top of heap while it has a higher priority than its parent
heapify_after_push( i, curId );
m_Stat.onPushSuccess();
*/
value_type * pop()
{
+ node& refTop = m_Heap[1];
+
m_Lock.lock();
if ( m_ItemCounter.value() == 0 ) {
// the heap is empty
m_Stat.onPopFailed();
return nullptr;
}
- counter_type nBottom = m_ItemCounter.reversed_value();
- m_ItemCounter.dec();
- // Since m_Heap[0] is not used, capacity() returns m_Heap.capacity() - 1
- // Consequently, "<=" is here
- assert( nBottom <= capacity() );
+ counter_type nBottom = m_ItemCounter.dec();
+ assert( nBottom < m_Heap.capacity() );
assert( nBottom > 0 );
- node& refBottom = m_Heap[ nBottom ];
+ if ( c_bFairQueue ) {
+ refTop.lock();
+ if ( nBottom == 1 ) {
+ refTop.m_nTag = tag_type( Empty );
+ value_type * pVal = refTop.m_pVal;
+ refTop.m_pVal = nullptr;
+ refTop.unlock();
+ m_Lock.unlock();
+ m_Stat.onPopSuccess();
+ return pVal;
+ }
+ }
+ node& refBottom = m_Heap[nBottom];
refBottom.lock();
m_Lock.unlock();
refBottom.m_nTag = tag_type(Empty);
refBottom.m_pVal = nullptr;
refBottom.unlock();
- node& refTop = m_Heap[ 1 ];
- refTop.lock();
+ //node& refTop = m_Heap[ 1 ];
+ if ( !c_bFairQueue )
+ refTop.lock();
+
if ( refTop.m_nTag == tag_type(Empty) ) {
// nBottom == nTop
refTop.unlock();
refTop.m_nTag = tag_type( Available );
// refTop will be unlocked inside heapify_after_pop
- heapify_after_pop( 1, &refTop );
+ heapify_after_pop( &refTop );
m_Stat.onPopSuccess();
return pVal;
template <typename Func>
void clear_with( Func f )
{
- while ( !empty() ) {
- value_type * pVal = pop();
- if ( pVal )
- f( *pVal );
- }
+ value_type * pVal;
+ while (( pVal = pop()) != nullptr )
+ f( *pVal );
}
/// Checks is the priority queue is empty
size_t size() const
{
std::unique_lock<lock_type> l( m_Lock );
- size_t nSize = (size_t) m_ItemCounter.value();
- return nSize;
+ return static_cast<size_t>( m_ItemCounter.value());
}
/// Return capacity of the priority queue
i = 0;
}
}
- else if ( refParent.m_nTag == tag_type(Empty) )
+ else if ( refParent.m_nTag == tag_type( Empty ) ) {
+ m_Stat.onItemMovedTop();
i = 0;
- else if ( refItem.m_nTag != curId )
+ }
+ else if ( refItem.m_nTag != curId ) {
+ m_Stat.onItemMovedUp();
i = nParent;
- else
+ }
+ else {
+ m_Stat.onPushEmptyPass();
bProgress = false;
+ }
refItem.unlock();
refParent.unlock();
}
}
- void heapify_after_pop( counter_type nParent, node * pParent )
+ void heapify_after_pop( node * pParent )
{
key_comparator cmp;
+ counter_type const nCapacity = m_Heap.capacity();
+
+ counter_type nParent = 1;
+ for ( counter_type nChild = nParent * 2; nChild < nCapacity; nChild *= 2 ) {
+ node* pChild = &m_Heap[ nChild ];
+ pChild->lock();
- while ( nParent < m_Heap.capacity() / 2 ) {
- counter_type nLeft = nParent * 2;
- counter_type nRight = nLeft + 1;
- node& refLeft = m_Heap[nLeft];
- node& refRight = m_Heap[nRight];
- refLeft.lock();
- refRight.lock();
-
- counter_type nChild;
- node * pChild;
- if ( refLeft.m_nTag == tag_type(Empty) ) {
- refRight.unlock();
- refLeft.unlock();
+ if ( pChild->m_nTag == tag_type( Empty )) {
+ pChild->unlock();
break;
}
- else if ( refRight.m_nTag == tag_type(Empty) || cmp( *refLeft.m_pVal, *refRight.m_pVal ) > 0 ) {
- refRight.unlock();
- nChild = nLeft;
- pChild = &refLeft;
- }
- else {
- refLeft.unlock();
- nChild = nRight;
- pChild = &refRight;
+
+ counter_type const nRight = nChild + 1;
+ if ( nRight < nCapacity ) {
+ node& refRight = m_Heap[nRight];
+ refRight.lock();
+
+ if ( refRight.m_nTag != tag_type( Empty ) && cmp( *refRight.m_pVal, *pChild->m_pVal ) > 0 ) {
+ // get right child
+ pChild->unlock();
+ nChild = nRight;
+ pChild = &refRight;
+ }
+ else
+ refRight.unlock();
}
- // If child has higher priority that parent then swap
+ // If child has higher priority than parent then swap
// Otherwise stop
if ( cmp( *pChild->m_pVal, *pParent->m_pVal ) > 0 ) {
std::swap( pParent->m_nTag, pChild->m_nTag );