schedule_backlog_tasks is now updated to fetch all tasks in the backlog_queue and then attempt to schedule them avoiding the infinite loop.

* Adding regression test `test_tiny_large_loop` that triggers `RecursionError: maximum recursion depth exceeded while getting the repr of an object`

Author: yadudoc

parsl/executors/high_throughput/mpi_resource_management.py

@@ -191,14 +191,19 @@ def put_task(self, task_package: dict):
 
     def _schedule_backlog_tasks(self):
         """Attempt to schedule backlogged tasks"""
-        try:
-            prioritized_task = self._backlog_queue.get(block=False)
-            self.put_task(prioritized_task.task)
-        except queue.Empty:
-            return
-        else:
-            # Keep attempting to schedule tasks till we are out of resources
-            self._schedule_backlog_tasks()
+
+        # Separate fetching tasks from the _backlog_queue and scheduling them
+        # since tasks that failed to schedule will be pushed to the _backlog_queue
+        backlogged_tasks = []
+        while True:
+            try:
+                prioritized_task = self._backlog_queue.get(block=False)
+                backlogged_tasks.append(prioritized_task.task)
+            except queue.Empty:
+                break
+
+        for backlogged_task in backlogged_tasks:
+            self.put_task(backlogged_task)
 
     def get_result(self, block: bool = True, timeout: Optional[float] = None):
         """Return result and relinquish provisioned nodes"""

parsl/tests/test_mpi_apps/test_mpi_scheduler.py

@@ -186,3 +186,35 @@ def test_hashable_backlog_queue():
         task_package = {"task_id": i, "buffer": mock_task_buffer}
         scheduler.put_task(task_package)
     assert scheduler._backlog_queue.qsize() == 2, "Expected 2 backlogged tasks"
+
+
+@pytest.mark.local
+def test_tiny_large_loop():
+    """Run a set of tiny and large tasks in a loop"""
+
+    task_q, result_q = SpawnContext.Queue(), SpawnContext.Queue()
+    scheduler = MPITaskScheduler(task_q, result_q)
+
+    assert scheduler.available_nodes
+    assert len(scheduler.available_nodes) == 8
+
+    assert scheduler._free_node_counter.value == 8
+
+    for i in range(10):
+        num_nodes = 2 if i % 2 == 0 else 8
+        mock_task_buffer = pack_res_spec_apply_message("func", "args", "kwargs",
+                                                       resource_specification={
+                                                           "num_nodes": num_nodes,
+                                                           "ranks_per_node": 2
+                                                       })
+        task_package = {"task_id": i, "buffer": mock_task_buffer}
+        scheduler.put_task(task_package)
+
+    for i in range(10):
+        task = task_q.get(timeout=30)
+        result_pkl = pickle.dumps(
+            {"task_id": task["task_id"], "type": "result", "buffer": "RESULT BUF"})
+        result_q.put(result_pkl)
+        got_result = scheduler.get_result(True, 1)
+
+    assert got_result == result_pkl