[core] Inject reorder_wait_seconds for scheduling queue test (ray-project#54404)

This scheduling queue unit test would take >30 seconds before because
the default value of reorder_wait_seconds is 30. Now just injecting it
in one level higher so can inject it into the test as 1 second.

Signed-off-by: dayshah <[email protected]>
Signed-off-by: ChanChan Mao <[email protected]>

Author: dayshah

src/ray/core_worker/test/scheduling_queue_test.cc

@@ -85,7 +85,7 @@ TEST(SchedulingQueueTest, TestTaskEvents) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   int n_ok = 0;
   int n_rej = 0;
   auto fn_ok = [&n_ok](const TaskSpecification &task_spec,
@@ -156,7 +156,7 @@ TEST(SchedulingQueueTest, TestInOrder) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   int n_ok = 0;
   int n_rej = 0;
   auto fn_ok = [&n_ok](const TaskSpecification &task_spec,
@@ -192,7 +192,7 @@ TEST(SchedulingQueueTest, TestWaitForObjects) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   std::atomic<int> n_ok(0);
   std::atomic<int> n_rej(0);
 
@@ -243,7 +243,7 @@ TEST(SchedulingQueueTest, TestWaitForObjectsNotSubjectToSeqTimeout) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   std::atomic<int> n_ok(0);
   std::atomic<int> n_rej(0);
 
@@ -286,7 +286,7 @@ TEST(SchedulingQueueTest, TestOutOfOrder) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   int n_ok = 0;
   int n_rej = 0;
   auto fn_ok = [&n_ok](const TaskSpecification &task_spec,
@@ -321,7 +321,7 @@ TEST(SchedulingQueueTest, TestSeqWaitTimeout) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   std::atomic<int> n_ok(0);
   std::atomic<int> n_rej(0);
 
@@ -337,7 +337,7 @@ TEST(SchedulingQueueTest, TestSeqWaitTimeout) {
   queue.Add(3, -1, fn_ok, fn_rej, nullptr, task_spec);
   ASSERT_TRUE(WaitForCondition(CreateEqualsConditionChecker(&n_ok, 1), 1000));
   ASSERT_EQ(n_rej, 0);
-  io_service.run();  // immediately triggers timeout
+  io_service.run();
   ASSERT_TRUE(WaitForCondition(CreateEqualsConditionChecker(&n_ok, 1), 1000));
   ASSERT_TRUE(WaitForCondition(CreateEqualsConditionChecker(&n_rej, 2), 1000));
   queue.Add(4, -1, fn_ok, fn_rej, nullptr, task_spec);
@@ -362,7 +362,7 @@ TEST(SchedulingQueueTest, TestSkipAlreadyProcessedByClient) {
   auto pool_manager =
       std::make_shared<ConcurrencyGroupManager<BoundedExecutor>>(concurrency_groups);
 
-  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager);
+  ActorSchedulingQueue queue(io_service, waiter, task_event_buffer, pool_manager, 1);
   std::atomic<int> n_ok(0);
   std::atomic<int> n_rej(0);
   auto fn_ok = [&n_ok](const TaskSpecification &task_spec,

src/ray/core_worker/transport/actor_scheduling_queue.cc

@@ -26,8 +26,10 @@ ActorSchedulingQueue::ActorSchedulingQueue(
     instrumented_io_context &task_execution_service,
     DependencyWaiter &waiter,
     worker::TaskEventBuffer &task_event_buffer,
-    std::shared_ptr<ConcurrencyGroupManager<BoundedExecutor>> pool_manager)
-    : wait_timer_(task_execution_service),
+    std::shared_ptr<ConcurrencyGroupManager<BoundedExecutor>> pool_manager,
+    int64_t reorder_wait_seconds)
+    : reorder_wait_seconds_(reorder_wait_seconds),
+      wait_timer_(task_execution_service),
       main_thread_id_(std::this_thread::get_id()),
       waiter_(waiter),
       task_event_buffer_(task_event_buffer),

src/ray/core_worker/transport/actor_scheduling_queue.h

@@ -45,7 +45,8 @@ class ActorSchedulingQueue : public SchedulingQueue {
       instrumented_io_context &task_execution_service,
       DependencyWaiter &waiter,
       worker::TaskEventBuffer &task_event_buffer,
-      std::shared_ptr<ConcurrencyGroupManager<BoundedExecutor>> pool_manager);
+      std::shared_ptr<ConcurrencyGroupManager<BoundedExecutor>> pool_manager,
+      int64_t reorder_wait_seconds);
 
   void Stop() override;
 
@@ -81,8 +82,7 @@ class ActorSchedulingQueue : public SchedulingQueue {
   /// Called when we time out waiting for an earlier task to show up.
   void OnSequencingWaitTimeout();
   /// Max time in seconds to wait for dependencies to show up.
-  const int64_t reorder_wait_seconds_ =
-      ::RayConfig::instance().actor_scheduling_queue_max_reorder_wait_seconds();
+  const int64_t reorder_wait_seconds_;
   /// Sorted map of (accept, rej) task callbacks keyed by their sequence number.
   std::map<int64_t, InboundRequest> pending_actor_tasks_;
   /// The next sequence number we are waiting for to arrive.

src/ray/core_worker/transport/task_receiver.cc

@@ -196,28 +196,29 @@ void TaskReceiver::HandleTask(rpc::PushTaskRequest request,
   if (task_spec.IsActorTask()) {
     auto it = actor_scheduling_queues_.find(task_spec.CallerWorkerId());
     if (it == actor_scheduling_queues_.end()) {
-      if (execute_out_of_order_) {
-        it = actor_scheduling_queues_
-                 .emplace(task_spec.CallerWorkerId(),
-                          std::make_unique<OutOfOrderActorSchedulingQueue>(
-                              task_execution_service_,
-                              waiter_,
-                              task_event_buffer_,
-                              pool_manager_,
-                              fiber_state_manager_,
-                              is_asyncio_,
-                              fiber_max_concurrency_,
-                              concurrency_groups_))
-                 .first;
-      } else {
-        it = actor_scheduling_queues_
-                 .emplace(task_spec.CallerWorkerId(),
-                          std::make_unique<ActorSchedulingQueue>(task_execution_service_,
-                                                                 waiter_,
-                                                                 task_event_buffer_,
-                                                                 pool_manager_))
-                 .first;
-      }
+      it = actor_scheduling_queues_
+               .emplace(
+                   task_spec.CallerWorkerId(),
+                   execute_out_of_order_
+                       ? std::unique_ptr<SchedulingQueue>(
+                             std::make_unique<OutOfOrderActorSchedulingQueue>(
+                                 task_execution_service_,
+                                 waiter_,
+                                 task_event_buffer_,
+                                 pool_manager_,
+                                 fiber_state_manager_,
+                                 is_asyncio_,
+                                 fiber_max_concurrency_,
+                                 concurrency_groups_))
+                       : std::unique_ptr<SchedulingQueue>(
+                             std::make_unique<ActorSchedulingQueue>(
+                                 task_execution_service_,
+                                 waiter_,
+                                 task_event_buffer_,
+                                 pool_manager_,
+                                 RayConfig::instance()
+                                     .actor_scheduling_queue_max_reorder_wait_seconds())))
+               .first;
     }
 
     it->second->Add(request.sequence_number(),