Hi @pengyk , thank you for this pull request! I have the following questions:

1. It's not very clear to me the purpose of Staging Queue. Could you please elaborate it a bit more?
2. I was thinking each worker only keeps a std::array<wsq_type, 3> _wsqs; where HIGH_PRIORITY=0, MEDIUM_PRIORITY=1, and LOW_PRIORITY=2, so we can simply iterate each of the three queues in order. In this case, steal and pop at the executor level will remain pretty much the same but with additional loop for priority first. pop is easier since we can just insert a node to the corresponding _wsqs[node->_priority].

I think perhaps the most efficient way to support priority is integrating it natively into work-stealing queue (WSQ). Supporting priority for unbounded WSQ might be a bit challenging now but I feel it is possible for bounded WSQ. Basically, we will maintain P priority splits on the ring buffer, where P can be 1 (default), 2, 4, etc. Each of the split works as a standalone WSQ in a single storage. We want P to be power of two so we can use efficient bit masking. In this case, the queue can natively support priority scheduling without any additional structure like std::array<wsq_type, 3> _wsqs;. Thought?

template <
  typename T, 
  size_t LogSize = TF_DEFAULT_BOUNDED_TASK_QUEUE_LOG_SIZE,
  size_t LogSplitSize = 0  // default 2^0 = 1, no priority support
> 
class BoundedWSQ {
  ... // we will support all priority-based push/pop/steal here.
}

Hi @tsung-wei-huang thanks for taking a look!

1. It is just to reduce the chances of a situation where, as worker is pushing its successors, another thread will not greedily steal a low priority task. It will "hide" the normal and low priority tasks and only expose the high priority tasks at first so that high priority tasks are guaranteed to be made visible first to be stolen. Imagine a graph like this

graph TD
    A[Root Node] --> B[Node 1<br/>Low priority]
    A --> C[Node 2<br/>Low priority]
    A --> D[Node 3<br/>Low priority]
    A --> E[Node 4<br/>High priority]
    A --> F[Node 5<br/>High priority]

If we have 3 threads and tasks are made ready in the given order, I think we can guarantee that the high priority tasks will all get executed first.

I do agree that this might be a little overkill and might degrade performance, so I am open to removing it.

I think the rest is pretty much just a std::array<wsq_type, 3> _wsqs except that I defined to be called _prio_wsq

2. This is a good idea! My one concern is that I think that right now we can execute multiple "topologies" on the same resources. So let's say we have one graph with 1 priority level, another with 2 and another with 8, wouldn't it be hard to keep track of which index is available for pushes? Also, would the semantics for defining the priority be harder to reason? Right now it is simply 3 hard coded enums for high/medium/low.

Let me know if that isn't clear, thanks!

@pengyk Thank you for the response! After a careful review, I suggest the following modification to refactor the pull:

1. Given the ultimate randomness of the work-stealing scheduler, I agree with you that Staging Queue is an overkill. The cost of its seems overshadows its benefit.
2. I would suggest let's focus on having priority-based scheduling at the worker level first, rather than the centralized spilling queue. The design of this centralized spilling queue is still being tested and evaluated in terms of its space and runtime tradeoff, and my require some future changes (e.g., NUMA adoption). On the other hand, the worker-level queue design is relatively fixed and stable. So let's focus on brining priority support to worker-level queue for now.

With 2, I suggest the following refactoring with a new class BoundedPriorityWSQ that abstract all the pop/steal/push operations with priority support so we can minimize the modification of tf::Executor

// let's define a concept for work-stealing queue
template <typename Q>
concept BoundedWSQLike = requires(Q& q) {
  { q.steal() };
  { q.pop() };
  { q.try_push() } -> bool ;  // unbounded wsq instead has push -> void
}

// top wrapper over WSQ - and yes, I agree with you that LogPriority is not a good idea. Let's just template it linearly
template <BoundedWSQLike Q, size_t MaxPriority>
class BoundedPriorityWSQ {
  
  public:
    // here we provide all BoundedWSQ methods with priority i to pin the operation to specific _wsq[i]
   template <typename O>   
   bool try_push(size_t priority, O&& item);

   ... similarly for other methods

   // for method that doesn't have priority, we will, by default, perform the batch operation from i=0, to MaxPriority-1   
   // note that i=0 means the highest priority
  template <typename O> 
  bool try_push(O&& item);
  
  ... similarly for other methods
   
  private:
  std::array<Q, MaxPriority> _wsqs;
}

Now, the advantage of this is, we can statically unroll all batch operations (e.g., BoundedPriorityWSQ::try_push(O&& item) in correct priority level.

// let's use the static unroll_until https://github.com/taskflow/taskflow/blob/a215b4578338cc27cba5c4772c4f73b61a538d42/taskflow/utility/traits.hpp#L164
template<auto beg, auto end, auto step, typename F>
constexpr bool unroll_until(F&& f) {
  return [&]<auto... Is>(std::index_sequence<Is...>) {
    return (f(beg + Is * step) || ...);
  }(std::make_index_sequence<(end - beg + step - 1) / step>{});
}


template <BoundedWSQLike Q, size_t MaxPriority>
template <typename O> 
bool BoundedPriorityWSQ<Q, MaxPriority>::try_push(O&& item) {
    return unroll_until<0, MaxPriority-1, 1>([&](size_t i){  return _wsqs[i].push(std::item); });
}

In this case, I don't think we even need prioritized_run? Thought?

Hi @tsung-wei-huang ,

Yeah that sounds good! I think this is a cleaner implementation. I am wondering if you have any preferences on the max number of MaxPriority. I think right now it is 3, is that ok or would you want it to be a configurable number?

@pengyk configurable number is definitely better. But we can default it to 3. Also, regarding setting the priority, instead of having a separate _priority member in tf::Node, we may reuse its _nstate like reserving 2 bits for priority so we can save some space (see NSTATEtaskflow/core/error.hpp).

template <BoundedWSQLike Q, size_t MaxPriority = 3>
template <typename O> 
bool BoundedPriorityWSQ<Q, MaxPriority>::try_push(O&& item) {
    return unroll_until<0, MaxPriority-1, 1>([&](size_t i){  return _wsqs[i].push(std::item); });
}

If you have any questions or any part that you think I can jump in or help, please don't hesitate to let me know.

@tsung-wei-huang Yeah that sounds good! I will try to get something out before end of week, thanks for the help! I will let you know if anything question comes up

Hi @tsung-wei-huang I think I have made the requested changes, let me know if anything is unclear thanks!

@pengyk thank you - I will have a look soon and get back to you! Before that, I will make a release first to include several other changes on profiler, doc examples, and other scheduler-level micro optimization. Really appreciate your contributions!

Hi @pengyk, I finally go time to look at this pull - thank you! Changes below:

1. I decouple the originally hard-coded priority() from the BoundedPriorityWSQ by making it a template. This allows applications to more generally return priority from their data type.
2. I assign the bit state 00 as high, 01 as normal, and 10 as low, consistent with your TaskPriority definition. This makes the priority return more straightforward (i.e., no extra remapping like queue_map = {0, 0, 1, 2}. By default, if users don't assign the priority, it is high, and the scanning in push/pop/steal always go from 0 to MaxPriority-1, naturally.
3. The biggest change is try_bulk_push which I delegate the push to individual try_bulk_push, rather than counting on executor to post process the return. The modification here performs a linear scan and calls try_bulk_push whenever the current priority is different from the previous one. first is naturally updated by individual try_bulk_push. When no priority is given at all, this function will bulk-push all nodes, at almost no extra cost.

  template <typename I>
  size_t try_bulk_push(I& first, size_t N) { 

    if(N == 0) return 0;

    size_t remaining = N; 

    while(remaining > 0) { 
      size_t q = _priority_fn(first[0]);
      size_t run_len = 1; 
      while(run_len < remaining && _priority_fn(first[run_len]) == q) { 
        ++run_len;
      }    
      size_t inserted = _wsqs[q].try_bulk_push(first, run_len);
      remaining -= inserted;
      if(inserted < run_len) {
        break;
      }    
    }    

    return N - remaining;
  }

With these changes, executor stays most of the same except the _update_cache function, which I slightly refactored it as follows:

  if(cache) {
    if(node->priority() < cache->priority()) {
      std::swap(cache, node);
    }    
    _schedule(worker, node);
  } else {
    cache = node;
  }

Please let me know your thought? Additionally, do you have any better suggestion for the naming BoundedPriorityWSQ? I think we also need unittest for BoundedPriorityWSQ. Would you like to help?

Hi @tsung-wei-huang thanks for looking!

The changes make sense to me. As for the renaming, maybe something like BoundedPriorityWSQSet to show that it is a multiple WSQs? or BoundedWSQSet directly?

Yeah I can work on the unit tests for that! Do you want it to be part of this PR or a separate one?

@pengyk Let's make it in this PR for consistency. You can add it to the existing work-stealing unittest file, unittests/test_wsq.cpp. Thank you!

Hi @tsung-wei-huang

I just added the unit tests, let me know if you want to change anything else, thanks!