tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn > Class Template Reference

class to create a lock-free bounded work-stealing queue with priority support More...

#include <taskflow/core/wsq.hpp>

Public Types
using	value_type = typename Q::value_type
	the return type of pop and steal operations

Public Member Functions
	BoundedPriorityWSQ (PriorityFn fn={})
	constructs the queue with an optional priority function
template<typename O>
bool	try_push (O &&item)
	tries to insert an item into the sub-queue determined by its priority
template<typename I>
size_t	try_bulk_push (I &first, size_t N)
	tries to insert a batch of items, routing contiguous same-priority runs to the corresponding sub-queue via bulk push
value_type	pop ()
	pops an item from the highest-priority non-empty sub-queue
value_type	steal ()
	steals an item from the highest-priority non-empty sub-queue
bool	empty () const noexcept
	queries whether all sub-queues are empty at the time of this call
size_t	size () const noexcept
	queries the total number of items across all sub-queues at the time of this call
constexpr size_t	capacity () const
	queries the total capacity across all sub-queues
Q &	operator[] (size_t priority)
	returns a reference to the sub-queue at the given priority level
const Q &	operator[] (size_t priority) const
	returns a const reference to the sub-queue at the given priority level

Detailed Description

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>
class tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >

class to create a lock-free bounded work-stealing queue with priority support

Template Parameters

Q	underlying bounded work-stealing queue type; must satisfy BoundedWSQLike
MaxPriority	number of priority levels (sub-queues); must be at least 1
PriorityFn	callable (const element&) -> size_t that maps an item to its priority index; the returned value must be in [0, MaxPriority). Defaults to DefaultPriorityFn.

BoundedPriorityWSQ wraps an array of MaxPriority sub-queues of type Q and routes each item to the sub-queue that corresponds to its priority. Priority levels are numbered from 0 to MaxPriority-1 in decreasing order of urgency: priority 0 is the highest and priority MaxPriority-1 is the lowest. Pop and steal operations always service sub-queues in ascending index order, so higher-priority items are consumed before lower-priority ones.

// 3-level priority queue with a custom priority function
auto priority_fn = [](tf::Node* n) -> size_t {
  // map application-level priority to [0, 3)
  return n->priority();  // must return 0, 1, or 2
};
 
tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3, decltype(priority_fn)> pq(priority_fn);
 
tf::Node* n = ...;
pq.try_push(n);
 
// pop always returns the highest-priority available item
auto item = pq.pop();

Member Typedef Documentation

value_type

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

using tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::value_type = typename Q::value_type

the return type of pop and steal operations

Inherits the value_type of the underlying queue Q. For pointer element types, it is the pointer type itself (using nullptr as the empty sentinel); for non-pointer types it is std::optional<T>.

Constructor & Destructor Documentation

BoundedPriorityWSQ()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::BoundedPriorityWSQ ( PriorityFn fn = {} )

inlineexplicit

constructs the queue with an optional priority function

Parameters

fn	priority function to use; defaults to a default-constructed PriorityFn

Member Function Documentation

capacity()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

size_t tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::capacity ( ) const

inlineconstexpr

queries the total capacity across all sub-queues

Returns

the capacity of one sub-queue multiplied by MaxPriority

// Each BoundedWSQ<Node*, 8> holds 256 items; 3 priority levels -> 768 total
tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
assert(pq.capacity() == 256 * 3);

empty()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

bool tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::empty ( ) const

inlinenoexcept

queries whether all sub-queues are empty at the time of this call

Returns

true if every sub-queue is empty, false otherwise

tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
assert(pq.empty() == true);
tf::Node* n = ...;
pq.try_push(n);
assert(pq.empty() == false);

operator[]() [1/2]

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

Q & tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::operator[] ( size_t priority )

inline

returns a reference to the sub-queue at the given priority level

Parameters

priority

priority index in [0, MaxPriority); 0 is the highest priority

tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
// directly access the high-priority sub-queue
auto item = pq[0].pop();
assert(pq[0].size() == 0);

operator[]() [2/2]

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

const Q & tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::operator[] ( size_t priority ) const

inline

returns a const reference to the sub-queue at the given priority level

Parameters

priority

priority index in [0, MaxPriority); 0 is the highest priority

const tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
assert(pq[0].empty() == true);

pop()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

value_type tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::pop ( )

inline

pops an item from the highest-priority non-empty sub-queue

Returns

the popped item, or empty_value() if all sub-queues are empty

Sub-queues are checked in ascending index order (index 0 is highest priority). The element is removed from the owner (back) end of the selected sub-queue, following last-in-first-out (LIFO) order within each priority level.

tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
// ... push items of various priorities ...
auto item = pq.pop();  // returns highest-priority available item
if(item != tf::BoundedWSQ<tf::Node*>::empty_value()) {
  // process item
}

Only the owner thread may call this method.

size()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

size_t tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::size ( ) const

inlinenoexcept

queries the total number of items across all sub-queues at the time of this call

Returns

sum of sizes of all MaxPriority sub-queues

tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
assert(pq.size() == 0);
tf::Node* n = ...;
pq.try_push(n);
assert(pq.size() == 1);

steal()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

value_type tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::steal ( )

inline

steals an item from the highest-priority non-empty sub-queue

Returns

the stolen item, or empty_value() if all sub-queues are empty

Sub-queues are checked in ascending index order (index 0 is highest priority). The element is removed from the thief (front) end of the selected sub-queue, following first-in-first-out (FIFO) order within each priority level.

tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3> pq;
// ... push items of various priorities ...
auto item = pq.steal();  // any thread may steal
if(item != tf::BoundedWSQ<tf::Node*>::empty_value()) {
  // process item
}

Any thread may call this method concurrently.

try_bulk_push()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

template<typename I>

size_t tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::try_bulk_push ( I & first, size_t N )

inline

tries to insert a batch of items, routing contiguous same-priority runs to the corresponding sub-queue via bulk push

Template Parameters

I	random-access iterator type

Parameters

first	iterator to the first item; advanced in place by the number of items inserted
N	number of items to insert starting at first

Returns

number of items successfully inserted

Scans first for contiguous runs of items that map to the same sub-queue (as determined by _priority_fn) and delegates each run to the underlying queue's try_bulk_push (a single atomic bottom update per run). Stops immediately if a run is only partially inserted (the target sub-queue became full), leaving first pointing at the first uninserted item.

auto fn = [](tf::Node* n) -> size_t { return n->priority(); };
tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3, decltype(fn)> pq(fn);
 
// Suppose items have priorities: 0 0 0 1 1 2 1 1 1 1 2 2
// Bulk push routes:
//   queue[0] <- 3 items (priority 0 run)
//   queue[1] <- 2 items (priority 1 run)
//   queue[2] <- 1 item  (priority 2 run)
//   queue[1] <- up to 4 items (priority 1 run); stops early if queue[1] is full
std::vector<tf::Node*> batch = ...;
auto it = batch.begin();
size_t inserted = pq.try_bulk_push(it, batch.size());
// it now points to batch.begin() + inserted

Only the owner thread may call this method.

try_push()

template<BoundedWSQLike Q, size_t MaxPriority, typename PriorityFn = DefaultPriorityFn>

template<typename O>

bool tf::BoundedPriorityWSQ< Q, MaxPriority, PriorityFn >::try_push ( O && item )

inline

tries to insert an item into the sub-queue determined by its priority

Template Parameters

O	item type (forwarded)

Parameters

item	the item to insert; _priority_fn(item) determines the target sub-queue and must return a value in [0, MaxPriority)

Returns

true if the item was inserted, false if the target sub-queue is full

auto fn = [](tf::Node* n) -> size_t { return n->priority(); };
tf::BoundedPriorityWSQ<tf::BoundedWSQ<tf::Node*>, 3, decltype(fn)> pq(fn);
tf::Node* n = ...;
bool ok = pq.try_push(n);

Only the owner thread may call this method.

---

The documentation for this class was generated from the following file:

• taskflow/core/wsq.hpp