Repository: triton-inference-server/server
Pull request: Add queue policy settings in dynamic batch scheduler. Add priority level for policy queues

Problem statement:
Server Queue
I have read the documentation and did not find any place talking about `Server Queue Size`. As far as I understood from the `TRTIS Architecture`, incoming inference requests are queued by `Model Schedulers` and when `Execution Context` is available, the request is passed for inference. I would like to know the Server Queue Size or if possible how to set it. This would help to control the incoming request traffic.


Requests support priority and timeout settings
1. In our online serving, some requests need a higher priority than others. For example, online request should have higher priority than offline request. Of course, the more priorities the better. This may be solved by either client with priority or request with priority.
2. It would be nice to have a request timeout parameter to achieve traffic control. For example, discard the request if its waiting time in queue exceeds the setting. This is especially important to avoid request backlog when serving large models.

Hints from issue discussion:
The scheduling queue doesn't have a fixed size, it grows to accommodate all pending requests. The latency for these requests will increase dramatically if the server is overloaded. Typically, something "upstream" of TRTIS should notice this increase in latency and scale / load-balance.

If TRTIS implemented a maximum queue size what should happen if the queue exceeds that size? Should requests just be rejected with "server overloaded" error? That is something that could be done.
I think it would be helpful to have "server overloaded" status or error, both would work. That way, the client could lower the request rate, or wait for the server for a while. 

It would be helpful to be able to configure queue size as well. For example, I could specify the queue size to be 1GB. Later, an API could be added to request server load status which would return load percentage (e.g. 74% loaded). 
Marked as enhancement. Can you use the latency of requests as a proxy for server overloaded? The latency will get very long when your queue starts to grow.
@deadeyegoodwin Thank you for the update. Yes, for now, I will monitor latency or will try to add alerts to the prometheus latency metrics.
Marking as enhancement request. We've had the "request timeout" feature requested before so we will look into prioritizing it.

I've just diagnosed a production incident where some combination of request deadlines (item 2 above) and queue control from https://github.com/NVIDIA/tensorrt-inference-server/issues/914 would have helped.

Specifically...

We have a separate cluster of TRTIS hosts behind an AWS load balancer servicing our "slow" models.  Each slow query can take a few seconds and pegs the GPU at 100%.  It looks like a burst of incoming calls caused requests to stack up and the TRTIS tasks ultimately hit out-of-memory.

Having a maximum queue size and a TRTIS per-request deadline would have helped mitigate the overload condition and prevented the OOMs too.
We're currently implementing a couple of "queuing policy" features and we will include priority, timeout and queue size.

Commit message: Fix FIXME

Patch to review:
diff --git a/src/backends/onnx/loader.cc b/src/backends/onnx/loader.cc
index 5847bdcdf9..3908913a3d 100644
--- a/src/backends/onnx/loader.cc
+++ b/src/backends/onnx/loader.cc
@@ -28,10 +28,10 @@
 
 #include <future>
 #include <thread>
-#include "src/core/logging.h"
 #include "src/backends/onnx/onnx_utils.h"
 #include "src/core/constants.h"
 #include "src/core/filesystem.h"
+#include "src/core/logging.h"
 
 namespace nvidia { namespace inferenceserver {
 
diff --git a/src/core/api.proto b/src/core/api.proto
index ef0633b673..52d08a2898 100644
--- a/src/core/api.proto
+++ b/src/core/api.proto
@@ -225,6 +225,27 @@ message InferRequestHeader
   //@@     request.
   //@@
   repeated Output output = 3;
+
+  //@@  .. cpp:var:: uint32 priority
+  //@@
+  //@@     The priority value of this request. If priority handling is not
+  //@@     enable for the model, then this value is ignored. The default value
+  //@@     is 0 which indicates that the request will be assigned the default
+  //@@     priority associated with the model.
+  //@@
+  uint32 priority = 7;
+
+  //@@  .. cpp:var:: uint64 timeout_microseconds
+  //@@
+  //@@     The timeout for this request. This value overrides the timeout
+  //@@     specified by the model, if the model allows timeout override and if
+  //@@     the value is less than the default timeout specified by the model.
+  //@@     If the request cannot be processed within this timeout, the request
+  //@@     will be handled based on the model's timeout policy.
+  //@@     The default value is 0 which indicates that the request does not
+  //@@     override the model's timeout value.
+  //@@
+  uint64 timeout_microseconds = 8;
 }
 
 //@@
diff --git a/src/core/backend.cc b/src/core/backend.cc
index a5cc3b537a..9f338fb2cd 100644
--- a/src/core/backend.cc
+++ b/src/core/backend.cc
@@ -97,6 +97,15 @@ InferenceBackend::SetModelConfig(
     }
   }
 
+  if (config_.has_dynamic_batching()) {
+    default_priority_level_ =
+        config_.dynamic_batching().default_priority_level();
+    max_priority_level_ = config_.dynamic_batching().priority_levels();
+  } else {
+    default_priority_level_ = 0;
+    max_priority_level_ = 0;
+  }
+
   return Status::Success;
 }
 
@@ -195,7 +204,10 @@ InferenceBackend::SetConfiguredScheduler(
         OnWarmup, OnRun, OnPeek, true /* dynamic_batching_enabled */,
         enforce_equal_shape_tensors,
         config_.dynamic_batching().preserve_ordering(), preferred_batch_sizes,
-        config_.dynamic_batching().max_queue_delay_microseconds(), &scheduler));
+        config_.dynamic_batching().max_queue_delay_microseconds(),
+        config_.dynamic_batching().default_queue_policy(),
+        config_.dynamic_batching().priority_levels(),
+        config_.dynamic_batching().priority_queue_policy(), &scheduler));
   } else {
     // Default scheduler. Use dynamic batch scheduler (with batching
     // disabled) as the default scheduler.
diff --git a/src/core/backend.h b/src/core/backend.h
index ae58ad95e9..092fd817f4 100644
--- a/src/core/backend.h
+++ b/src/core/backend.h
@@ -90,6 +90,10 @@ class InferenceBackend {
       const std::shared_ptr<InferResponseProvider>& response_provider,
       std::function<void(const Status&)> OnCompleteHandleInfer);
 
+  uint32_t DefaultPriorityLevel() const { return default_priority_level_; }
+
+  uint32_t MaxPriorityLevel() const { return max_priority_level_; }
+
  protected:
   // Run model on the context associated with 'runner_idx' to
   // execute for one or more requests.
@@ -163,6 +167,12 @@ class InferenceBackend {
 
   // Path to model
   std::string model_dir_;
+
+  // The default priority level for the backend.
+  uint32_t default_priority_level_;
+
+  // The largest priority value for the backend.
+  uint32_t max_priority_level_;
 };
 
 }}  // namespace nvidia::inferenceserver
diff --git a/src/core/dynamic_batch_scheduler.cc b/src/core/dynamic_batch_scheduler.cc
index 1924831f64..3d1c8c733a 100644
--- a/src/core/dynamic_batch_scheduler.cc
+++ b/src/core/dynamic_batch_scheduler.cc
@@ -49,14 +49,17 @@ DynamicBatchScheduler::DynamicBatchScheduler(
     const std::unordered_map<std::string, bool>& enforce_equal_shape_tensors,
     const bool preserve_ordering,
     const std::set<int32_t>& preferred_batch_sizes,
-    const uint64_t max_queue_delay_microseconds)
+    const uint64_t max_queue_delay_microseconds,
+    const ModelQueuePolicy& default_queue_policy,
+    const uint32_t priority_levels, const ModelQueuePolicyMap& queue_policy_map)
     : OnInit_(OnInit), OnWarmup_(OnWarmup), OnSchedule_(OnSchedule),
       OnPeek_(OnPeek), dynamic_batching_enabled_(dynamic_batching_enabled),
       scheduler_thread_cnt_(runner_cnt), idle_scheduler_thread_cnt_(0),
+      queue_(default_queue_policy, priority_levels, queue_policy_map),
       preferred_batch_sizes_(preferred_batch_sizes),
       pending_batch_delay_ns_(max_queue_delay_microseconds * 1000),
-      pending_batch_size_(0), pending_batch_queue_cnt_(0),
-      queued_batch_size_(0), next_preferred_batch_size_(0),
+      pending_batch_size_(0), queued_batch_size_(0),
+      next_preferred_batch_size_(0),
       enforce_equal_shape_tensors_(enforce_equal_shape_tensors),
       preserve_ordering_(preserve_ordering)
 {
@@ -79,11 +82,34 @@ DynamicBatchScheduler::Create(
     const std::set<int32_t>& preferred_batch_sizes,
     const uint64_t max_queue_delay_microseconds,
     std::unique_ptr<Scheduler>* scheduler)
+{
+  return Create(
+      runner_id_start, runner_cnt, nice, OnInit, OnWarmup, OnSchedule, OnPeek,
+      dynamic_batching_enabled, enforce_equal_shape_tensors, preserve_ordering,
+      preferred_batch_sizes, max_queue_delay_microseconds, ModelQueuePolicy(),
+      0, ModelQueuePolicyMap(), scheduler);
+}
+
+Status
+DynamicBatchScheduler::Create(
+    const uint32_t runner_id_start, const uint32_t runner_cnt, const int nice,
+    const StandardInitFunc& OnInit, const StandardWarmupFunc& OnWarmup,
+    const StandardRunFunc& OnSchedule,
+    const StandardShapeTensorPeekFunc& OnPeek,
+    const bool dynamic_batching_enabled,
+    const std::unordered_map<std::string, bool>& enforce_equal_shape_tensors,
+    const bool preserve_ordering,
+    const std::set<int32_t>& preferred_batch_sizes,
+    const uint64_t max_queue_delay_microseconds,
+    const ModelQueuePolicy& default_queue_policy,
+    const uint32_t priority_levels, const ModelQueuePolicyMap& queue_policy_map,
+    std::unique_ptr<Scheduler>* scheduler)
 {
   DynamicBatchScheduler* dyna_sched = new DynamicBatchScheduler(
       runner_id_start, runner_cnt, OnInit, OnWarmup, OnSchedule, OnPeek,
       dynamic_batching_enabled, enforce_equal_shape_tensors, preserve_ordering,
-      preferred_batch_sizes, max_queue_delay_microseconds);
+      preferred_batch_sizes, max_queue_delay_microseconds, default_queue_policy,
+      priority_levels, queue_policy_map);
   std::unique_ptr<DynamicBatchScheduler> sched(dyna_sched);
 
   // Create one scheduler thread for each requested runner. Associate
@@ -162,11 +188,18 @@ DynamicBatchScheduler::Enqueue(
   // scheduling process
   stats->CaptureTimestamp(ModelInferStats::TimestampKind::kQueueStart);
 
+  const auto& request = request_provider->Request();
+  Status enqueue_status;
   bool wake_runner = false;
   {
     std::lock_guard<std::mutex> lock(mu_);
-    queue_.emplace_back(stats, request_provider, response_provider, OnComplete);
-    queued_batch_size_ += request_provider->Request()->BatchSize();
+    enqueue_status = queue_.Enqueue(
+        request->Priority(),
+        std::move(
+            Payload(stats, request_provider, response_provider, OnComplete)));
+    if (enqueue_status.IsOk()) {
+      queued_batch_size_ += request->BatchSize();
+    }
 
     // If there are any idle runners and the queued batch size is greater or
     // equal to next preferred batch size, then wake one up to service this
@@ -184,6 +217,10 @@ DynamicBatchScheduler::Enqueue(
   if (wake_runner) {
     cv_.notify_one();
   }
+
+  if (!enqueue_status.IsOk()) {
+    OnComplete(enqueue_status);
+  }
 }
 
 void
@@ -254,6 +291,8 @@ DynamicBatchScheduler::SchedulerThread(
     NVTX_RANGE(nvtx_, "DynamicBatchScheduler " + runner_id);
 
     std::shared_ptr<std::vector<Scheduler::Payload>> payloads;
+    std::shared_ptr<std::vector<std::deque<Scheduler::Payload>>>
+        rejected_payloads;
     bool wake_thread = false;
     uint64_t wait_microseconds = 0;
 
@@ -264,22 +303,27 @@ DynamicBatchScheduler::SchedulerThread(
         // Debugging/testing... wait until queue contains 'delay_cnt'
         // items...
         wait_microseconds = 10 * 1000;
-        if (queue_.size() >= delay_cnt) {
+        if (queue_.Size() >= delay_cnt) {
           delay_cnt = 0;
         }
         LOG_INFO << "Delaying scheduler thread " << runner_id << " until "
                  << delay_cnt
-                 << " queued payloads, current total = " << queue_.size();
-      } else if (queue_.empty()) {
+                 << " queued payloads, current total = " << queue_.Size();
+      } else if (queue_.Empty()) {
         wait_microseconds = default_wait_microseconds;
       } else if (dynamic_batching_enabled_) {
         // Use dynamic batching to get request payload(s) to execute.
         wait_microseconds = GetDynamicBatch(runner_id);
-        if (wait_microseconds == 0) {
+
+        // Get payloads that are rejected from searching dynamic batch.
+        rejected_payloads = queue_.ReleaseRejectedPayloads();
+
+        // Extract batch only if there is pending batch
+        auto pending_batch_queue_cnt = queue_.PendingBatchCount();
+        if ((wait_microseconds == 0) && (pending_batch_queue_cnt != 0)) {
           payloads = std::make_shared<std::vector<Scheduler::Payload>>();
-          for (size_t idx = 0; idx < pending_batch_queue_cnt_; ++idx) {
-            payloads->emplace_back(std::move(queue_.front()));
-            queue_.pop_front();
+          for (size_t idx = 0; idx < pending_batch_queue_cnt; ++idx) {
+            payloads->emplace_back(std::move(queue_.Dequeue()));
           }
           if (preserve_ordering_) {
             std::lock_guard<std::mutex> lock(completion_id_queue_mtx_);
@@ -295,7 +339,6 @@ DynamicBatchScheduler::SchedulerThread(
           next_preferred_batch_size_ = 0;
 
           pending_batch_size_ = 0;
-          pending_batch_queue_cnt_ = 0;
           pending_batch_shapes_.clear();
 
           // If there are still requests in the queue after removing
@@ -307,13 +350,12 @@ DynamicBatchScheduler::SchedulerThread(
           // handling those requests. We do the actual wake outside of
           // the lock to avoid having the woken thread immediately
           // block on the lock.
-          wake_thread = !queue_.empty() && (idle_scheduler_thread_cnt_ > 0);
+          wake_thread = !queue_.Empty() && (idle_scheduler_thread_cnt_ > 0);
         }
       } else {
         // No batching... execute next request payload
         payloads = std::make_shared<std::vector<Scheduler::Payload>>();
-        payloads->emplace_back(std::move(queue_.front()));
-        queue_.pop_front();
+        payloads->emplace_back(std::move(queue_.Dequeue()));
         if (preserve_ordering_) {
           std::lock_guard<std::mutex> lock(completion_id_queue_mtx_);
           completion_id_queue_.push(completion_id);
@@ -351,6 +393,19 @@ DynamicBatchScheduler::SchedulerThread(
       }
     }
 
+    // Finish rejected payloads if any
+    if (rejected_payloads != nullptr) {
+      static Status rejected_status =
+          Status(RequestStatusCode::UNAVAILABLE, "Request timeout expired");
+      for (auto& rejected_queue : *rejected_payloads) {
+        for (auto& rejected_payload : rejected_queue) {
+          if (rejected_payload.complete_function_ != nullptr) {
+            rejected_payload.complete_function_(rejected_status);
+          }
+        }
+      }
+    }
+
     // At the end of this scope 'payloads' will be destroyed.  A
     // handle to the backend is held through the
     // payload.complete_function_. If the server is exiting or the
@@ -381,22 +436,24 @@ DynamicBatchScheduler::GetDynamicBatch(const int64_t runner_id)
   // batch size would be exceeded or if the shape of the next request
   // does not match the shape of the pending batch.
   bool send_now = false;
+  if (!queue_.IsCursorValid()) {
+    queue_.ResetCursor();
+    pending_batch_size_ = 0;
+  }
   size_t best_preferred_batch_size = 0;
-  size_t best_preferred_batch_cnt = 0;
-  size_t search_batch_size = pending_batch_size_;
-  size_t search_batch_cnt = pending_batch_queue_cnt_;
-  for (auto idx = pending_batch_queue_cnt_; idx < queue_.size(); ++idx) {
+  queued_batch_size_ -= queue_.ApplyPolicyAtCursor();
+  while (!queue_.CursorEnd()) {
     const auto batch_size =
-        queue_[idx].request_provider_->Request()->BatchSize();
+        queue_.PayloadAtCursor().request_provider_->Request()->BatchSize();
 
     // If there is no pending batch, then this request is starting a
     // new batch.
-    if (search_batch_cnt == 0) {
+    if (queue_.PendingBatchCount() == 0) {
       // Get the shape of the new batch that is being started...
       if (!enforce_equal_shape_tensors_.empty()) {
         if (!InitPendingShape(
-                 runner_id, queue_[idx], enforce_equal_shape_tensors_, OnPeek_,
-                 &pending_batch_shapes_)
+                 runner_id, queue_.PayloadAtCursor(),
+                 enforce_equal_shape_tensors_, OnPeek_, &pending_batch_shapes_)
                  .IsOk()) {
           send_now = true;
           break;
@@ -405,7 +462,7 @@ DynamicBatchScheduler::GetDynamicBatch(const int64_t runner_id)
     } else {
       // There is a pending batch and adding this request would make
       // the batch size too large, so send the pending batch as it is.
-      if ((search_batch_size + batch_size) > max_preferred_batch_size_) {
+      if ((pending_batch_size_ + batch_size) > max_preferred_batch_size_) {
         send_now = true;
         break;
       }
@@ -414,36 +471,34 @@ DynamicBatchScheduler::GetDynamicBatch(const int64_t runner_id)
       // this request, so send the pending batch as it is.
       if (!enforce_equal_shape_tensors_.empty() &&
           !CompareWithPendingShape(
-              runner_id, queue_[idx], OnPeek_, pending_batch_shapes_)) {
+              runner_id, queue_.PayloadAtCursor(), OnPeek_,
+              pending_batch_shapes_)) {
         send_now = true;
         break;
       }
     }
 
-    search_batch_size += batch_size;
-    search_batch_cnt++;
+    pending_batch_size_ += batch_size;
+    queue_.AdvanceCursor();
+    queued_batch_size_ -= queue_.ApplyPolicyAtCursor();
 
-    if (preferred_batch_sizes_.find(search_batch_size) !=
+    if (preferred_batch_sizes_.find(pending_batch_size_) !=
         preferred_batch_sizes_.end()) {
-      best_preferred_batch_size = search_batch_size;
-      best_preferred_batch_cnt = search_batch_cnt;
+      best_preferred_batch_size = pending_batch_size_;
+      queue_.MarkCursor();
     }
   }
 
   // If we found a preferred batch size then execute that.
   if (best_preferred_batch_size != 0) {
     pending_batch_size_ = best_preferred_batch_size;
-    pending_batch_queue_cnt_ = best_preferred_batch_cnt;
+    queue_.SetCursorToMark();
     return 0;
   }
 
-  pending_batch_size_ = search_batch_size;
-  pending_batch_queue_cnt_ = search_batch_cnt;
-
-  // Should always have at least one request in the pending batch at
-  // this point.
-  if (pending_batch_queue_cnt_ == 0) {
-    LOG_ERROR << "unexpected pending batch size 0";
+  // No request in pending batch happens when all queued requests have expired
+  // timeout and the policies are REJECT
+  if (queue_.PendingBatchCount() == 0) {
     return 0;
   }
 
@@ -461,9 +516,7 @@ DynamicBatchScheduler::GetDynamicBatch(const int64_t runner_id)
   // a thread to check again at the maximum allowed delay.
   struct timespec now;
   clock_gettime(CLOCK_MONOTONIC, &now);
-  const struct timespec& queued = queue_.front().stats_->Timestamp(
-      ModelInferStats::TimestampKind::kQueueStart);
-  uint64_t delay_ns = TIMESPEC_TO_NANOS(now) - TIMESPEC_TO_NANOS(queued);
+  uint64_t delay_ns = TIMESPEC_TO_NANOS(now) - queue_.OldestEnqueueTime();
 
   if (delay_ns >= pending_batch_delay_ns_) {
     return 0;
diff --git a/src/core/dynamic_batch_scheduler.h b/src/core/dynamic_batch_scheduler.h
index 51895071ab..9100bc2ad6 100644
--- a/src/core/dynamic_batch_scheduler.h
+++ b/src/core/dynamic_batch_scheduler.h
@@ -29,6 +29,7 @@
 #include <condition_variable>
 #include <deque>
 #include <future>
+#include <map>
 #include <mutex>
 #include <queue>
 #include <set>
@@ -59,6 +60,24 @@ class DynamicBatchScheduler : public Scheduler {
       const uint64_t max_queue_delay_microseconds,
       std::unique_ptr<Scheduler>* scheduler);
 
+  // Create a scheduler to support a given number of runners and a run
+  // function to call when a request is scheduled. And the scheduler also
+  // supports different queue policies for different priority levels.
+  static Status Create(
+      const uint32_t runner_id_start, const uint32_t runner_cnt, const int nice,
+      const StandardInitFunc& OnInit, const StandardWarmupFunc& OnWarmup,
+      const StandardRunFunc& OnSchedule,
+      const StandardShapeTensorPeekFunc& OnPeek,
+      const bool dynamic_batching_enabled,
+      const std::unordered_map<std::string, bool>& enforce_equal_shape_tensors,
+      const bool preserve_ordering,
+      const std::set<int32_t>& preferred_batch_sizes,
+      const uint64_t max_queue_delay_microseconds,
+      const ModelQueuePolicy& default_queue_policy,
+      const uint32_t priority_level,
+      const ModelQueuePolicyMap& queue_policy_map,
+      std::unique_ptr<Scheduler>* scheduler);
+
   ~DynamicBatchScheduler();
 
   // \see Scheduler::Enqueue()
@@ -78,7 +97,10 @@ class DynamicBatchScheduler : public Scheduler {
       const std::unordered_map<std::string, bool>& enforce_equal_shape_tensors,
       const bool preserve_ordering,
       const std::set<int32_t>& preferred_batch_sizes,
-      const uint64_t max_queue_delay_microseconds);
+      const uint64_t max_queue_delay_microseconds,
+      const ModelQueuePolicy& default_queue_policy,
+      const uint32_t priority_levels,
+      const ModelQueuePolicyMap& queue_policy_map);
   void SchedulerThread(
       const uint32_t runner_id, const uint32_t completion_id, const int nice,
       const std::shared_ptr<std::atomic<bool>>& rthread_exit,
@@ -115,9 +137,10 @@ class DynamicBatchScheduler : public Scheduler {
   std::mutex mu_;
   std::condition_variable cv_;
 
-  // Queue holding inference requests for the model represented by
-  // this scheduler.
-  std::deque<Scheduler::Payload> queue_;
+  // Map from priority level to queue holding inference requests for the model
+  // represented by this scheduler. If priority queues are not supported by the
+  // scheduler, then priority zero entry is used as the single queue?
+  PriorityQueue queue_;
 
   std::vector<std::unique_ptr<std::thread>> scheduler_threads_;
   std::vector<std::shared_ptr<std::atomic<bool>>> scheduler_threads_exit_;
@@ -126,7 +149,6 @@ class DynamicBatchScheduler : public Scheduler {
   std::set<int32_t> preferred_batch_sizes_;
   uint64_t pending_batch_delay_ns_;
   size_t pending_batch_size_;
-  size_t pending_batch_queue_cnt_;
   PendingBatchShapes pending_batch_shapes_;
 
   size_t queued_batch_size_;
diff --git a/src/core/infer_request.cc b/src/core/infer_request.cc
index f8efbb07ea..02960193cf 100644
--- a/src/core/infer_request.cc
+++ b/src/core/infer_request.cc
@@ -177,6 +177,10 @@ InferenceRequest::Normalize(const InferenceBackend& backend)
 
   const ModelConfig& model_config = backend.Config();
 
+  if ((priority_ == 0) || (priority_ > backend.MaxPriorityLevel())) {
+    priority_ = backend.DefaultPriorityLevel();
+  }
+
   // FIXMEV2 For V2 protocol we must adjust the shape of the input
   // tensors to remove the batch dimension and instead report that as
   // batch-size.
diff --git a/src/core/infer_request.h b/src/core/infer_request.h
index e2785d6ec6..5e1a54a2cc 100644
--- a/src/core/infer_request.h
+++ b/src/core/infer_request.h
@@ -223,6 +223,12 @@ class InferenceRequest {
   uint32_t BatchSize() const { return batch_size_; }
   void SetBatchSize(uint32_t b) { batch_size_ = b; }
 
+  uint32_t Priority() const { return priority_; }
+  void SetPriority(uint32_t p) { priority_ = p; }
+
+  uint64_t TimeoutMs() const { return timeout_ms_; }
+  void SetTimeoutMs(uint64_t t) { timeout_ms_ = t; }
+
   std::unordered_map<std::string, Input>* MutableInputs() { return &inputs_; }
   const std::unordered_map<std::string, Input>& Inputs() const
   {
@@ -283,6 +289,8 @@ class InferenceRequest {
   uint32_t flags_;
   uint64_t correlation_id_;
   uint32_t batch_size_;
+  uint32_t priority_;
+  uint64_t timeout_ms_;
 
   std::unordered_map<std::string, Input> inputs_;
   std::unordered_map<std::string, RequestedOutput> requested_outputs_;
diff --git a/src/core/model_config.proto b/src/core/model_config.proto
index b3ec648d1b..24d82161d2 100644
--- a/src/core/model_config.proto
+++ b/src/core/model_config.proto
@@ -623,6 +623,71 @@ message ModelOptimizationPolicy
   PinnedMemoryBuffer output_pinned_memory = 6;
 }
 
+//@@
+//@@.. cpp:var:: message ModelQueuePolicy
+//@@
+//@@   Queue policy for inference requests.
+//@@
+message ModelQueuePolicy
+{
+  //@@
+  //@@  .. cpp:enum:: TimeoutAction
+  //@@
+  //@@     The action applied on timeout request.
+  //@@
+  enum TimeoutAction {
+    //@@    .. cpp:enumerator:: Action::REJECT = 0
+    //@@
+    //@@       Reject the request and return error message accordingly.
+    //@@
+    REJECT = 0;
+
+    //@@    .. cpp:enumerator:: Action::DELAY = 1
+    //@@
+    //@@       Delay the request until all other requests at the same
+    //@@       (or higher) priority levels that have not reached their timeouts
+    //@@       are processed. A delayed request will eventually be processed,
+    //@@       but may be delayed indefinitely due to newly arriving requests.
+    //@@
+    DELAY = 1;
+  }
+
+  //@@
+  //@@  .. cpp:var:: TimeoutAction timeout_action
+  //@@
+  //@@     The action applied on timeout request. The default action is REJECT.
+  //@@
+  TimeoutAction timeout_action = 1;
+
+  //@@
+  //@@  .. cpp:var:: uint64 default_timeout_microseconds
+  //@@
+  //@@     The default timeout for every request, in microseconds.
+  //@@     The default value is 0 which indicates that no timeout is set.
+  //@@
+  uint64 default_timeout_microseconds = 2;
+
+  //@@
+  //@@  .. cpp:var:: bool allow_timeout_override
+  //@@
+  //@@     Whether individual request can override the default timeout value.
+  //@@     When true, individual requests can set a timeout that is less than
+  //@@     the default timeout value but may not increase the timeout.
+  //@@     The default value is false.
+  //@@
+  bool allow_timeout_override = 3;
+
+  //@@
+  //@@  .. cpp:var:: uint32 max_queue_size
+  //@@
+  //@@     The maximum queue size for holding requests. A request will be
+  //@@     rejected immediately if it can't be enqueued because the queue is
+  //@@     full. The default value is 0 which indicates that no maximum
+  //@@     queue size is enforced.
+  //@@
+  uint32 max_queue_size = 4;
+}
+
 //@@
 //@@.. cpp:var:: message ModelDynamicBatching
 //@@
@@ -661,6 +726,40 @@ message ModelDynamicBatching
   //@@     order.
   //@@
   bool preserve_ordering = 3;
+
+  //@@  .. cpp:var:: uint32 priority_levels
+  //@@
+  //@@     The number of priority levels to be enabled for the model,
+  //@@     the priority level starts from 1. Requests with lower priority level
+  //@@     will be handled before requests with higher priority levels.
+  //@@     Requests with the same priority level will be handled in the order
+  //@@     that they are received.
+  //@@
+  uint32 priority_levels = 4;
+
+  //@@  .. cpp:var:: uint32 default_priority_level
+  //@@
+  //@@     The priority level used for requests that don't specify their
+  //@@     priority. The value must be in the range [ 1, 'priority_levels' ].
+  //@@
+  uint32 default_priority_level = 5;
+
+  //@@  .. cpp:var:: ModelQueuePolicy default_queue_policy
+  //@@
+  //@@     The default queue policy used for requests that don't require
+  //@@     priority handling and requests that specify priority levels where
+  //@@     there is no specific policy given. If not specified, a policy with
+  //@@     default field values will be used.
+  //@@
+  ModelQueuePolicy default_queue_policy = 6;
+
+  //@@  .. cpp:var:: map<uint32, ModelQueuePolicy> priority_queue_policy
+  //@@
+  //@@     Specify the queue policy for the priority level. The default queue
+  //@@     policy will be used if a priority level doesn't specify a queue
+  //@@     policy.
+  //@@
+  map<uint32, ModelQueuePolicy> priority_queue_policy = 7;
 }
 
 //@@
diff --git a/src/core/model_config_utils.cc b/src/core/model_config_utils.cc
index 62141301c7..d94796e21e 100644
--- a/src/core/model_config_utils.cc
+++ b/src/core/model_config_utils.cc
@@ -512,6 +512,29 @@ ValidateModelConfig(
                 config.name());
       }
     }
+
+    // Priority queue is specified
+    const auto priority_levels = config.dynamic_batching().priority_levels();
+    if (priority_levels != 0) {
+      if ((config.dynamic_batching().default_priority_level() == 0) ||
+          (config.dynamic_batching().default_priority_level() >
+           priority_levels)) {
+        return Status(
+            RequestStatusCode::INVALID_ARG,
+            "default priority level must be in range [1, " +
+                std::to_string(priority_levels) + "]");
+      }
+      for (const auto& queue_policy :
+           config.dynamic_batching().priority_queue_policy()) {
+        if ((queue_policy.first == 0) ||
+            (queue_policy.first > priority_levels)) {
+          return Status(
+              RequestStatusCode::INVALID_ARG,
+              "priority queue policy must have priority level in range [1, " +
+                  std::to_string(priority_levels) + "]");
+        }
+      }
+    }
   }
 
   // If sequence batching is specified make sure the control is
diff --git a/src/core/scheduler.h b/src/core/scheduler.h
index 71126925f8..a95b1ec112 100644
--- a/src/core/scheduler.h
+++ b/src/core/scheduler.h
@@ -65,6 +65,16 @@ class Scheduler {
     {
     }
 
+    Payload& operator=(Payload&& payload)
+    {
+      stats_ = std::move(payload.stats_);
+      request_provider_ = std::move(payload.request_provider_);
+      response_provider_ = std::move(payload.response_provider_);
+      complete_function_ = std::move(payload.complete_function_);
+      status_ = payload.status_;
+      return *this;
+    }
+
     std::shared_ptr<ModelInferStats> stats_;
     std::shared_ptr<InferRequestProvider> request_provider_;
     std::shared_ptr<InferResponseProvider> response_provider_;
diff --git a/src/core/scheduler_utils.cc b/src/core/scheduler_utils.cc
index 74200a4f81..515904cff2 100644
--- a/src/core/scheduler_utils.cc
+++ b/src/core/scheduler_utils.cc
@@ -26,6 +26,8 @@
 
 #include "src/core/scheduler_utils.h"
 
+#include <cassert>
+#include "src/core/constants.h"
 #include "src/core/provider.h"
 
 namespace nvidia { namespace inferenceserver {
@@ -97,4 +99,282 @@ CompareWithPendingShape(
   return true;
 }
 
+Status
+PriorityQueue::PolicyQueue::Enqueue(Scheduler::Payload&& payload)
+{
+  if ((max_queue_size_ != 0) && (Size() >= max_queue_size_)) {
+    return Status(RequestStatusCode::UNAVAILABLE, "Exceeds maximum queue size");
+  }
+  queue_.emplace_back(std::move(payload));
+  auto timeout_ms = default_timeout_ms_;
+  if (allow_timeout_override_) {
+    auto override_timeout_ms =
+        queue_.back().request_provider_->Request()->TimeoutMs();
+    if (override_timeout_ms != 0 && override_timeout_ms < timeout_ms) {
+      timeout_ms = override_timeout_ms;
+    }
+  }
+  if (timeout_ms != 0) {
+    struct timespec now;
+    clock_gettime(CLOCK_MONOTONIC, &now);
+    timeout_timestamp_ns_.emplace_back(
+        TIMESPEC_TO_NANOS(now) + timeout_ms * 1000);
+  } else {
+    timeout_timestamp_ns_.emplace_back(0);
+  }
+
+  return Status::Success;
+}
+
+Scheduler::Payload
+PriorityQueue::PolicyQueue::Dequeue()
+{
+  if (!queue_.empty()) {
+    auto res = std::move(queue_.front());
+    queue_.pop_front();
+    timeout_timestamp_ns_.pop_front();
+    return res;
+  } else {
+    auto res = std::move(delayed_queue_.front());
+    delayed_queue_.pop_front();
+    return res;
+  }
+}
+
+bool
+PriorityQueue::PolicyQueue::ApplyPolicy(
+    size_t idx, size_t* rejected_count, size_t* rejected_batch_size)
+{
+  struct timespec now;
+  clock_gettime(CLOCK_MONOTONIC, &now);
+  auto now_nanoseconds = TIMESPEC_TO_NANOS(now);
+  if (idx < queue_.size()) {
+    size_t curr_idx = idx;
+    while (curr_idx < queue_.size()) {
+      if ((timeout_timestamp_ns_[curr_idx] != 0) &&
+          (now_nanoseconds > timeout_timestamp_ns_[curr_idx])) {
+        if (timeout_action_ == ModelQueuePolicy::DELAY) {
+          delayed_queue_.emplace_back(std::move(queue_[curr_idx]));
+        } else {
+          rejected_queue_.emplace_back(std::move(queue_[curr_idx]));
+          *rejected_count += 1;
+          *rejected_batch_size += rejected_queue_.back()
+                                      .request_provider_->Request()
+                                      ->BatchSize();
+        }
+        curr_idx++;
+      }
+      break;
+    }
+
+    // Use range erasure on deque as all erasure functions are linear,
+    // this implies in the edge case where this function is always called on
+    // 'bad' index can be O(n^2). However, for data structures that are O(1)
+    // erasure, the traversal may not be as efficient due to cache miss
+    // (elements not stored contiguously).
+    queue_.erase(queue_.begin() + idx, queue_.begin() + curr_idx);
+    timeout_timestamp_ns_.erase(
+        timeout_timestamp_ns_.begin() + idx,
+        timeout_timestamp_ns_.begin() + curr_idx);
+
+    // Current idx is pointing to an item with unexpired timeout
+    if (idx < queue_.size()) {
+      return true;
+    }
+  }
+  // At this point, idx is pointing to an item with expired timeout.
+  // If the item is in delayed queue, then return true. Otherwise, false
+  // meaning the queue has no item with this 'idx'.
+  return ((idx - queue_.size()) < delayed_queue_.size());
+}
+
+std::deque<Scheduler::Payload>
+PriorityQueue::PolicyQueue::ReleaseRejectedQueue()
+{
+  std::deque<Scheduler::Payload> res;
+  rejected_queue_.swap(res);
+  return res;
+}
+
+Scheduler::Payload&
+PriorityQueue::PolicyQueue::At(size_t idx)
+{
+  if (idx < queue_.size()) {
+    return queue_[idx];
+  } else {
+    return delayed_queue_[idx - queue_.size()];
+  }
+}
+
+uint64_t
+PriorityQueue::PolicyQueue::TimeoutAt(size_t idx)
+{
+  if (idx < queue_.size()) {
+    return timeout_timestamp_ns_[idx];
+  } else {
+    return 0;
+  }
+}
+
+PriorityQueue::PriorityQueue()
+    : size_(0), front_priority_level_(0), last_priority_level_(0)
+{
+  ModelQueuePolicy default_policy;
+  queues_.emplace(0, PolicyQueue(default_policy));
+  front_priority_level_ = queues_.begin()->first;
+  ResetCursor();
+}
+
+PriorityQueue::PriorityQueue(
+    const ModelQueuePolicy& default_queue_policy, uint32_t priority_levels,
+    const ModelQueuePolicyMap queue_policy_map)
+    : size_(0), last_priority_level_(priority_levels)
+{
+  if (priority_levels == 0) {
+    queues_.emplace(0, PolicyQueue(default_queue_policy));
+  } else {
+    for (uint32_t level = 1; level <= priority_levels; level++) {
+      auto it = queue_policy_map.find(level);
+      if (it == queue_policy_map.end()) {
+        queues_.emplace(level, PolicyQueue(default_queue_policy));
+      } else {
+        queues_.emplace(level, PolicyQueue(it->second));
+      }
+    }
+  }
+  front_priority_level_ = queues_.begin()->first;
+  ResetCursor();
+}
+
+Status
+PriorityQueue::Enqueue(uint32_t priority_level, Scheduler::Payload&& payload)
+{
+  auto status = queues_[priority_level].Enqueue(std::move(payload));
+  if (status.IsOk()) {
+    size_++;
+    front_priority_level_ = std::min(front_priority_level_, priority_level);
+    // Invalidate the pending batch cursor if the enqueued item is placed
+    // in within the pending batch. At the same priority level, the payload is
+    // guaranteed to be after pending batch if the batch hasn't reached
+    // delayed queue.
+    if ((priority_level < pending_cursor_.curr_it_->first) ||
+        ((priority_level == pending_cursor_.curr_it_->first) &&
+         (pending_cursor_.at_delayed_queue_))) {
+      pending_cursor_.valid_ = false;
+    }
+  }
+  return status;
+}
+
+Scheduler::Payload
+PriorityQueue::Dequeue()
+{
+  pending_cursor_.valid_ = false;
+  if (!queues_[front_priority_level_].Empty()) {
+    size_--;
+    return queues_[front_priority_level_].Dequeue();
+  } else if (front_priority_level_ != last_priority_level_) {
+    front_priority_level_++;
+  }
+  throw std::out_of_range("dequeue on empty queue");
+}
+
+std::shared_ptr<std::vector<std::deque<Scheduler::Payload>>>
+PriorityQueue::ReleaseRejectedPayloads()
+{
+  auto res = std::make_shared<std::vector<std::deque<Scheduler::Payload>>>(
+      queues_.size());
+  size_t idx = 0;
+  for (auto& queue : queues_) {
+    (*res)[idx] = std::move(queue.second.ReleaseRejectedQueue());
+    idx++;
+  }
+  return std::move(res);
+}
+
+bool
+PriorityQueue::IsCursorValid()
+{
+  if (pending_cursor_.valid_) {
+    struct timespec now;
+    clock_gettime(CLOCK_MONOTONIC, &now);
+    return TIMESPEC_TO_NANOS(now) <
+           pending_cursor_.pending_batch_closest_timeout_ns_;
+  }
+  return false;
+}
+
+PriorityQueue::Cursor::Cursor(PriorityQueues::iterator start_it)
+    : curr_it_(start_it), queue_idx_(0), at_delayed_queue_(false),
+      pending_batch_closest_timeout_ns_(0),
+      pending_batch_oldest_enqueue_time_ns_(0), pending_batch_count_(0),
+      valid_(true)
+{
+}
+
+size_t
+PriorityQueue::ApplyPolicyAtCursor()
+{
+  size_t rejected_batch_size = 0;
+  size_t rejected_count = 0;
+  while (pending_cursor_.curr_it_ != queues_.end()) {
+    if (!(pending_cursor_.curr_it_->second.ApplyPolicy(
+            pending_cursor_.queue_idx_, &rejected_count,
+            &rejected_batch_size))) {
+      if (size_ > pending_cursor_.pending_batch_count_ + rejected_count) {
+        pending_cursor_.curr_it_++;
+        pending_cursor_.queue_idx_ = 0;
+        continue;
+      }
+    }
+    // Control reach here if the cursor points to a payload that is candidate
+    // for pending batch, or if all payloads are in pending batch.
+    break;
+  }
+  // DEBUG: remove the check below. We don't want cursor to go beyond 'queues_'
+  // so that we can simply call ApplyPolicyAtCursor() again when new payloads
+  // are added, and we can make sure that curr_it_ can be dereferenced safely.
+  assert(pending_cursor_.curr_it_ != queues_.end());
+  size_ -= rejected_count;
+  return rejected_batch_size;
+}
+
+void
+PriorityQueue::AdvanceCursor()
+{
+  if (pending_cursor_.pending_batch_count_ >= size_) {
+    return;
+  }
+
+  const auto& timeout_ns =
+      pending_cursor_.curr_it_->second.TimeoutAt(pending_cursor_.queue_idx_);
+  if (timeout_ns != 0) {
+    if (pending_cursor_.pending_batch_closest_timeout_ns_ != 0) {
+      pending_cursor_.pending_batch_closest_timeout_ns_ = std::min(
+          pending_cursor_.pending_batch_closest_timeout_ns_, timeout_ns);
+    } else {
+      pending_cursor_.pending_batch_closest_timeout_ns_ = timeout_ns;
+    }
+  }
+
+  auto curr_enqueue_time_ns = TIMESPEC_TO_NANOS(
+      pending_cursor_.curr_it_->second.At(pending_cursor_.queue_idx_)
+          .stats_->Timestamp(ModelInferStats::TimestampKind::kQueueStart));
+  if (pending_cursor_.pending_batch_oldest_enqueue_time_ns_ != 0) {
+    pending_cursor_.pending_batch_oldest_enqueue_time_ns_ = std::min(
+        pending_cursor_.pending_batch_oldest_enqueue_time_ns_,
+        curr_enqueue_time_ns);
+  } else {
+    pending_cursor_.pending_batch_oldest_enqueue_time_ns_ =
+        curr_enqueue_time_ns;
+  }
+  ++pending_cursor_.queue_idx_;
+  ++pending_cursor_.pending_batch_count_;
+  // pending batch includes delayed payload if (queue_idx_ - 1) points to
+  // delayed queue.
+  pending_cursor_.at_delayed_queue_ =
+      (pending_cursor_.queue_idx_ >
+       pending_cursor_.curr_it_->second.UnexpiredSize());
+}
+
 }}  // namespace nvidia::inferenceserver
diff --git a/src/core/scheduler_utils.h b/src/core/scheduler_utils.h
index 9a1080d7ac..9db5fb9ad4 100644
--- a/src/core/scheduler_utils.h
+++ b/src/core/scheduler_utils.h
@@ -25,6 +25,7 @@
 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #pragma once
 
+#include <deque>
 #include <unordered_map>
 #include "src/core/model_config.h"
 #include "src/core/scheduler.h"
@@ -46,4 +47,190 @@ bool CompareWithPendingShape(
     const Scheduler::StandardShapeTensorPeekFunc& OnPeek,
     const PendingBatchShapes& pending_batch_shapes);
 
+using ModelQueuePolicyMap =
+    ::google::protobuf::Map<::google::protobuf::uint32, ModelQueuePolicy>;
+
+class PriorityQueue {
+ public:
+  // Construct a queue with no priority level with default queue policy,
+  // which will behave the same as regular queue.
+  PriorityQueue();
+
+  // Construct a queue with 'priority_levels', the priority starts from 1.
+  // Different priority level may follow different queue policies given by
+  // 'queue_policy_map', otherwise, the 'default_queue_policy' will be used.
+  PriorityQueue(
+      const ModelQueuePolicy& default_queue_policy, uint32_t priority_levels,
+      const ModelQueuePolicyMap queue_policy_map);
+
+  // Enqueue 'payload' with priority set to 'priority_level'.
+  Status Enqueue(uint32_t priority_level, Scheduler::Payload&& payload);
+
+  // Dequeue the payload at the front of the queue.
+  Scheduler::Payload Dequeue();
+
+  // Retrieve the payloads that are rejected based on the queue policies.
+  std::shared_ptr<std::vector<std::deque<Scheduler::Payload>>>
+  ReleaseRejectedPayloads();
+
+  // Return the number of payloads in the queue, rejected payloads are not
+  // included.
+  size_t Size() { return size_; }
+
+  // Whether the queue is empty, rejected payloads are not included.
+  bool Empty() { return Size() == 0; }
+
+  // Reset the cursor such that it is representing an empty pending batch.
+  void ResetCursor() { pending_cursor_ = Cursor(queues_.begin()); }
+
+  // Record the current cursor. The cursor can be restored to recorded state
+  // by invoking SetCursorToMark(). Note that Enqueue(), Dequeue(), and
+  // ResetCursor() will invalidate the marker, it is the function caller's
+  // responsibility to ensure the marker is valid before calling
+  // SetCursorToMark().
+  void MarkCursor() { current_mark_ = pending_cursor_; }
+
+  // Apply the queue policy and alter the underlying queue accordingly. After
+  // the function returns, the cursor may be at its end to indicate that
+  // there no request after the pending batch.
+  // Returns the total batch size of the newly rejected requests.
+  size_t ApplyPolicyAtCursor();
+
+  // Return the payload at cursor.
+  Scheduler::Payload& PayloadAtCursor()
+  {
+    return pending_cursor_.curr_it_->second.At(pending_cursor_.queue_idx_);
+  }
+
+  // Advance the cursor for pending batch. This function will not trigger the
+  // queue policy. No effect if the cursor already reach the end of the queue.
+  void AdvanceCursor();
+
+  // Whether the cursor reaches its end,
+  bool CursorEnd() { return pending_cursor_.pending_batch_count_ == size_; }
+
+  // Restore the cursor state to the marker.
+  void SetCursorToMark() { pending_cursor_ = current_mark_; }
+
+  // Whether the cursor is still valid. The cursor is valid only if the pending
+  // batch is unchanged.
+  bool IsCursorValid();
+
+  // Return the oldest queued time of payloads in pending batch.
+  uint64_t OldestEnqueueTime()
+  {
+    return pending_cursor_.pending_batch_oldest_enqueue_time_ns_;
+  }
+
+  // Return the number of payloads in pending batch.
+  size_t PendingBatchCount() { return pending_cursor_.pending_batch_count_; }
+
+ private:
+  class PolicyQueue {
+   public:
+    // Construct a policy queue with default policy, which will behave the same
+    // as regular queue.
+    PolicyQueue()
+        : timeout_action_(ModelQueuePolicy::REJECT), default_timeout_ms_(0),
+          allow_timeout_override_(false), max_queue_size_(0)
+    {
+    }
+
+    // Construct a policy queue with given 'policy'.
+    PolicyQueue(const ModelQueuePolicy& policy)
+        : timeout_action_(policy.timeout_action()),
+          default_timeout_ms_(policy.default_timeout_microseconds()),
+          allow_timeout_override_(policy.allow_timeout_override()),
+          max_queue_size_(policy.max_queue_size())
+    {
+    }
+
+    // Enqueue an payload and set up its timeout accordingly.
+    Status Enqueue(Scheduler::Payload&& payload);
+
+    // Dequeue the payload at the front of the queue.
+    Scheduler::Payload Dequeue();
+
+    // Apply the queue policy to payload at 'idx'.
+    // 'rejected_count' will be incremented by the number of the newly rejected
+    // requets after applying the policy.
+    // 'rejected_batch_size' will be incremented by the total batch size of the
+    // newly rejected requets after applying the policy.
+    // Return true if the 'idx' still points to an payload after applying the
+    // policy, false otherwise.
+    bool ApplyPolicy(
+        size_t idx, size_t* rejected_count, size_t* rejected_batch_size);
+
+    // Return the rejected payloads held by the request queue.
+    std::deque<Scheduler::Payload> ReleaseRejectedQueue();
+
+    // Return the payload at 'idx'.
+    Scheduler::Payload& At(size_t idx);
+
+    // Return the timeout timestamp of the payload at 'idx', in ns. A value of 0
+    // indicates that the payload doesn't specify a timeout.
+    uint64_t TimeoutAt(size_t idx);
+
+    // Return whether the queue is empty, rejected requests are not included.
+    bool Empty() { return Size() == 0; }
+
+    // Return the number of requests in the queue, rejected requests are not
+    // included.
+    size_t Size() { return queue_.size() + delayed_queue_.size(); }
+
+    // Return the number of unexpired requests in the queue
+    size_t UnexpiredSize() { return queue_.size(); }
+
+   private:
+    // Variables that define the policy for the queue
+    const ModelQueuePolicy::TimeoutAction timeout_action_;
+    const uint64_t default_timeout_ms_;
+    const bool allow_timeout_override_;
+    const uint32_t max_queue_size_;
+
+    std::deque<uint64_t> timeout_timestamp_ns_;
+    std::deque<Scheduler::Payload> queue_;
+    std::deque<Scheduler::Payload> delayed_queue_;
+    std::deque<Scheduler::Payload> rejected_queue_;
+  };
+  using PriorityQueues = std::map<uint32_t, PolicyQueue>;
+
+  // Cursor for tracking pending batch, the cursor points to the item after
+  // the pending batch.
+  struct Cursor {
+    Cursor() = default;
+    Cursor(PriorityQueues::iterator start_it);
+
+    Cursor(const Cursor& rhs)
+        : curr_it_(rhs.curr_it_), queue_idx_(rhs.queue_idx_),
+          at_delayed_queue_(rhs.at_delayed_queue_),
+          pending_batch_closest_timeout_ns_(
+              rhs.pending_batch_closest_timeout_ns_),
+          pending_batch_oldest_enqueue_time_ns_(
+              rhs.pending_batch_oldest_enqueue_time_ns_),
+          pending_batch_count_(rhs.pending_batch_count_), valid_(rhs.valid_)
+    {
+    }
+
+    PriorityQueues::iterator curr_it_;
+    size_t queue_idx_;
+    bool at_delayed_queue_;
+    uint64_t pending_batch_closest_timeout_ns_;
+    uint64_t pending_batch_oldest_enqueue_time_ns_;
+    size_t pending_batch_count_;
+    bool valid_;
+  };
+
+  PriorityQueues queues_;
+  size_t size_;
+
+  // Keep track of the priority level that the first payload in the queue
+  // is at to avoid traversing 'queues_'
+  uint32_t front_priority_level_;
+  uint32_t last_priority_level_;
+
+  Cursor pending_cursor_;
+  Cursor current_mark_;
+};
+
 }}  // namespace nvidia::inferenceserver


Code review comments:
diff_hunk: @@ -97,4 +99,282 @@ CompareWithPendingShape(
   return true;
 }
 
+Status
+PriorityQueue::PolicyQueue::Enqueue(Scheduler::Payload&& payload)
+{
+  if ((max_queue_size_ != 0) && (Size() >= max_queue_size_)) {
+    return Status(RequestStatusCode::UNAVAILABLE, "Exceeds maximum queue size");
+  }
+  queue_.emplace_back(std::move(payload));
+  auto timeout_ms = default_timeout_ms_;
+  if (allow_timeout_override_) {
+    auto override_timeout_ms =
+        queue_.back().request_provider_->Request()->TimeoutMs();
+    if (override_timeout_ms != 0 && override_timeout_ms < timeout_ms) {
+      timeout_ms = override_timeout_ms;
+    }
+  }
+  if (timeout_ms != 0) {
+    struct timespec now;
+    clock_gettime(CLOCK_MONOTONIC, &now);
+    timeout_timestamp_ns_.emplace_back(
+        TIMESPEC_TO_NANOS(now) + timeout_ms * 1000);
+  } else {
+    timeout_timestamp_ns_.emplace_back(0);
+  }
+
+  return Status::Success;
+}
+
+Scheduler::Payload
+PriorityQueue::PolicyQueue::Dequeue()
+{
+  if (!queue_.empty()) {
+    auto res = std::move(queue_.front());
+    queue_.pop_front();
+    timeout_timestamp_ns_.pop_front();
+    return res;
+  } else {
+    auto res = std::move(delayed_queue_.front());
+    delayed_queue_.pop_front();
+    return res;
+  }
+}
+
+bool
+PriorityQueue::PolicyQueue::ApplyPolicy(
+    size_t idx, size_t* rejected_count, size_t* rejected_batch_size)
+{
+  struct timespec now;
+  clock_gettime(CLOCK_MONOTONIC, &now);
+  auto now_nanoseconds = TIMESPEC_TO_NANOS(now);
+  if (idx < queue_.size()) {
+    size_t curr_idx = idx;
+    while (curr_idx < queue_.size()) {
+      if ((timeout_timestamp_ns_[curr_idx] != 0) &&
+          (now_nanoseconds > timeout_timestamp_ns_[curr_idx])) {
+        if (timeout_action_ == ModelQueuePolicy::DELAY) {
+          delayed_queue_.emplace_back(std::move(queue_[curr_idx]));
+        } else {
+          rejected_queue_.emplace_back(std::move(queue_[curr_idx]));
+          *rejected_count += 1;
+          *rejected_batch_size += rejected_queue_.back()
+                                      .request_provider_->Request()
+                                      ->BatchSize();
+        }
+        curr_idx++;
+      }
+      break;
+    }
+
+    // Use range erasure on deque as all erasure functions are linear,
+    // this implies in the edge case where this function is always called on
+    // 'bad' index can be O(n^2). However, for data structures that are O(1)
+    // erasure, the traversal may not be as efficient due to cache miss
+    // (elements not stored contiguously).
+    queue_.erase(queue_.begin() + idx, queue_.begin() + curr_idx);
+    timeout_timestamp_ns_.erase(
+        timeout_timestamp_ns_.begin() + idx,
+        timeout_timestamp_ns_.begin() + curr_idx);
+
+    // Current idx is pointing to an item with unexpired timeout
+    if (idx < queue_.size()) {
+      return true;
+    }
+  }
+  // At this point, idx is pointing to an item with expired timeout.
+  // If the item is in delayed queue, then return true. Otherwise, false
+  // meaning the queue has no item with this 'idx'.
+  return ((idx - queue_.size()) < delayed_queue_.size());
+}
+
+std::deque<Scheduler::Payload>
+PriorityQueue::PolicyQueue::ReleaseRejectedQueue()
+{
+  std::deque<Scheduler::Payload> res;
+  rejected_queue_.swap(res);
+  return res;
+}
+
+Scheduler::Payload&
+PriorityQueue::PolicyQueue::At(size_t idx)
+{
+  if (idx < queue_.size()) {
+    return queue_[idx];
+  } else {
+    return delayed_queue_[idx - queue_.size()];
+  }
+}
+
+uint64_t
+PriorityQueue::PolicyQueue::TimeoutAt(size_t idx)
+{
+  if (idx < queue_.size()) {
+    return timeout_timestamp_ns_[idx];
+  } else {
+    return 0;
+  }
+}
+
+PriorityQueue::PriorityQueue()
+    : size_(0), front_priority_level_(0), last_priority_level_(0)
+{
+  ModelQueuePolicy default_policy;
+  queues_.emplace(0, PolicyQueue(default_policy));
+  front_priority_level_ = queues_.begin()->first;
+  ResetCursor();
+}
+
+PriorityQueue::PriorityQueue(
+    const ModelQueuePolicy& default_queue_policy, uint32_t priority_levels,
+    const ModelQueuePolicyMap queue_policy_map)
+    : size_(0), last_priority_level_(priority_levels)
+{
+  if (priority_levels == 0) {
+    queues_.emplace(0, PolicyQueue(default_queue_policy));
+  } else {
+    for (uint32_t level = 1; level <= priority_levels; level++) {
+      auto it = queue_policy_map.find(level);
+      if (it == queue_policy_map.end()) {
+        queues_.emplace(level, PolicyQueue(default_queue_policy));
+      } else {
+        queues_.emplace(level, PolicyQueue(it->second));
+      }
+    }
+  }
+  front_priority_level_ = queues_.begin()->first;
+  ResetCursor();
+}
+
+Status
+PriorityQueue::Enqueue(uint32_t priority_level, Scheduler::Payload&& payload)
+{
+  auto status = queues_[priority_level].Enqueue(std::move(payload));
+  if (status.IsOk()) {
+    size_++;
+    front_priority_level_ = std::min(front_priority_level_, priority_level);
+    // Invalidate the pending batch cursor if the enqueued item is placed
+    // in within the pending batch. At the same priority level, the payload is
path: src/core/scheduler_utils.cc
text: @user1:
Remove "in"

diff_hunk: @@ -223,6 +223,12 @@ class InferenceRequest {
   uint32_t BatchSize() const { return batch_size_; }
   void SetBatchSize(uint32_t b) { batch_size_ = b; }
 
+  uint32_t Priority() const { return priority_; }
+  void SetPriority(uint32_t p) { priority_ = p; }
+
+  uint64_t TimeoutMs() const { return timeout_ms_; }
path: src/core/infer_request.h
text: @user1:
TimeoutMicroseconds and change variable to timeout_microseconds_ or timeout_us_

diff_hunk: @@ -115,9 +137,10 @@ class DynamicBatchScheduler : public Scheduler {
   std::mutex mu_;
   std::condition_variable cv_;
 
-  // Queue holding inference requests for the model represented by
-  // this scheduler.
-  std::deque<Scheduler::Payload> queue_;
+  // Map from priority level to queue holding inference requests for the model
+  // represented by this scheduler. If priority queues are not supported by the
+  // scheduler, then priority zero entry is used as the single queue?
path: src/core/dynamic_batch_scheduler.h
text: @user1:
Remove "?"